Design a trading app with real time charts and trades.

package com.androidinterview.trading.book;

import com.androidinterview.trading.feed.MarketMessage.BookSnapshot;
import com.androidinterview.trading.feed.MarketMessage.Level;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.TreeMap;

// Depth for one symbol, as two sorted maps. Bids descend so the best bid is
// first, asks ascend so the best ask is first, and the top of each side is a
// first key rather than a scan.
//
// This class is deliberately dumb about ordering. It applies what it is given.
// The feed above it is what guarantees nothing was skipped.
public final class OrderBook {

    private final NavigableMap<Long, Long> bids = new TreeMap<>(Collections.reverseOrder());
    private final NavigableMap<Long, Long> asks = new TreeMap<>();

    // A snapshot replaces the book. It does not merge into it, because the
    // whole reason a snapshot arrived is that the old contents are suspect.
    public void reset(BookSnapshot snapshot) {
        bids.clear();
        asks.clear();
        for (Level level : snapshot.bids()) put(bids, level);
        for (Level level : snapshot.asks()) put(asks, level);
    }

    public void apply(boolean bid, Level level) {
        put(bid ? bids : asks, level);
    }

    // Zero quantity means the level is gone, so it is removed rather than
    // stored. Keeping empty levels leaves rows of nothing at the top of the
    // ladder and quietly breaks the best bid and best ask.
    private static void put(NavigableMap<Long, Long> side, Level level) {
        if (level.quantity() == 0) side.remove(level.priceTicks());
        else side.put(level.priceTicks(), level.quantity());
    }

    public Long bestBid() {
        return bids.isEmpty() ? null : bids.firstKey();
    }

    public Long bestAsk() {
        return asks.isEmpty() ? null : asks.firstKey();
    }

    // The screen shows ten rows a side, not the four thousand levels the feed
    // knows about, so the read is bounded even when the book is not.
    public List<Level> top(boolean bid, int depth) {
        List<Level> out = new ArrayList<>(depth);
        for (Map.Entry<Long, Long> entry : (bid ? bids : asks).entrySet()) {
            if (out.size() == depth) break;
            out.add(new Level(entry.getKey(), entry.getValue()));
        }
        return out;
    }
}

package com.androidinterview.trading.book;

import com.androidinterview.trading.feed.MarketMessage.BookSnapshot;
import com.androidinterview.trading.feed.MarketMessage.Level;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.TreeMap;

public final class OrderBook {

    private final NavigableMap<Long, Long> bids = new TreeMap<>(Collections.reverseOrder());
    private final NavigableMap<Long, Long> asks = new TreeMap<>();

    public void reset(BookSnapshot snapshot) {
        bids.clear();
        asks.clear();
        for (Level level : snapshot.bids()) put(bids, level);
        for (Level level : snapshot.asks()) put(asks, level);
    }

    public void apply(boolean bid, Level level) {
        put(bid ? bids : asks, level);
    }

    private static void put(NavigableMap<Long, Long> side, Level level) {
        if (level.quantity() == 0) side.remove(level.priceTicks());
        else side.put(level.priceTicks(), level.quantity());
    }

    public Long bestBid() {
        return bids.isEmpty() ? null : bids.firstKey();
    }

    public Long bestAsk() {
        return asks.isEmpty() ? null : asks.firstKey();
    }

    public List<Level> top(boolean bid, int depth) {
        List<Level> out = new ArrayList<>(depth);
        for (Map.Entry<Long, Long> entry : (bid ? bids : asks).entrySet()) {
            if (out.size() == depth) break;
            out.add(new Level(entry.getKey(), entry.getValue()));
        }
        return out;
    }
}

package com.androidinterview.trading.chart;

import com.androidinterview.trading.feed.MarketMessage.Trade;

// Where a thousand ticks a second becomes sixty frames a second.
//
// The socket thread folds every trade into the live candle and writes the
// result into one field. The UI thread reads that field once a frame and draws
// it. Nothing is queued, nothing is posted, and no tick is ever handed to the
// main thread, so the cost of a fast feed is a little arithmetic on a
// background thread rather than a dropped frame.
//
// Historical candles come from REST, once, and are immutable. Only the newest
// candle is built here, because only the newest one is still changing.
public final class CandleBuilder {

    public record Candle(long openMillis, long open, long high, long low, long close, long volume) {

        Candle fold(long priceTicks, long quantity) {
            return new Candle(
                    openMillis,
                    open,
                    Math.max(high, priceTicks),
                    Math.min(low, priceTicks),
                    priceTicks,
                    volume + quantity);
        }
    }

    private final long intervalMillis;

    // Volatile, so the UI thread reads a fully published object rather than a
    // half written one. A candle is immutable, so a sample is always coherent
    // even when the next tick lands mid draw.
    private volatile Candle live;

    // A diagnostic, so a test can show how many ticks one sample stood for.
    private long foldedTicks;

    public CandleBuilder(long intervalMillis) {
        if (intervalMillis <= 0) throw new IllegalArgumentException("interval must be positive");
        this.intervalMillis = intervalMillis;
    }

    // Called on the socket thread, thousands of times a second. Returns the
    // candle that just closed, so the caller can append it to the history, or
    // null when the live candle simply grew.
    public Candle fold(Trade trade) {
        foldedTicks++;
        // The bucket comes from the exchange's clock. A phone five minutes
        // fast would file this trade in the wrong minute and draw a chart that
        // disagrees with every other screen in the world.
        long bucket = trade.serverMillis() - Math.floorMod(trade.serverMillis(), intervalMillis);
        Candle current = live;

        if (current == null || bucket > current.openMillis()) {
            long price = trade.priceTicks();
            live = new Candle(bucket, price, price, price, price, trade.quantity());
            return current;
        }
        // A trade stamped before the live candle opened arrived late. It
        // belongs to a candle already sent to history, so it is dropped rather
        // than allowed to rewrite the past.
        if (bucket < current.openMillis()) return null;

        live = current.fold(trade.priceTicks(), trade.quantity());
        return null;
    }

    // Called on the UI thread, at most once a frame. Every tick between two
    // calls is already folded into what this returns, which is the whole
    // point, the intermediate states were never worth drawing.
    public Candle sample() {
        return live;
    }

    public long foldedTicks() {
        return foldedTicks;
    }
}

package com.androidinterview.trading.chart;

import com.androidinterview.trading.feed.MarketMessage.Trade;

public final class CandleBuilder {

    public record Candle(long openMillis, long open, long high, long low, long close, long volume) {

        Candle fold(long priceTicks, long quantity) {
            return new Candle(
                    openMillis,
                    open,
                    Math.max(high, priceTicks),
                    Math.min(low, priceTicks),
                    priceTicks,
                    volume + quantity);
        }
    }

    private final long intervalMillis;

    private volatile Candle live;

    private long foldedTicks;

    public CandleBuilder(long intervalMillis) {
        if (intervalMillis <= 0) throw new IllegalArgumentException("interval must be positive");
        this.intervalMillis = intervalMillis;
    }

    public Candle fold(Trade trade) {
        foldedTicks++;
        long bucket = trade.serverMillis() - Math.floorMod(trade.serverMillis(), intervalMillis);
        Candle current = live;

        if (current == null || bucket > current.openMillis()) {
            long price = trade.priceTicks();
            live = new Candle(bucket, price, price, price, price, trade.quantity());
            return current;
        }
        if (bucket < current.openMillis()) return null;

        live = current.fold(trade.priceTicks(), trade.quantity());
        return null;
    }

    public Candle sample() {
        return live;
    }

    public long foldedTicks() {
        return foldedTicks;
    }
}

package com.androidinterview.trading.connection;

import java.util.function.LongUnaryOperator;

// The lifecycle of the market data socket, as five states the UI can render
// honestly.
//
// The failure this exists to prevent is the silent one. A socket that stops
// delivering looks exactly like a market that stopped moving, and a trader who
// cannot tell those apart will place an order against a price from four
// minutes ago.
public final class FeedConnection {

    public enum State {
        CONNECTING, LIVE,
        // Still connected, but nothing has arrived for longer than it should
        // have. The prices stay on screen with their server timestamp beside
        // them, greyed, rather than pretending to be current.
        STALE,
        RECONNECTING,
        // Backgrounded. Holding a market data socket behind a locked screen
        // burns battery to render nothing, so it closes and a resume takes a
        // fresh snapshot.
        CLOSED
    }

    private final long stalenessMillis;
    private final long baseBackoffMillis;
    private final long maxBackoffMillis;
    private final LongUnaryOperator jitter;

    private State state = State.CONNECTING;
    private int attempt;
    private long lastMessageElapsed;
    private long lastServerMillis;

    public FeedConnection(long stalenessMillis, long baseBackoffMillis, long maxBackoffMillis, LongUnaryOperator jitter) {
        this.stalenessMillis = stalenessMillis;
        this.baseBackoffMillis = baseBackoffMillis;
        this.maxBackoffMillis = maxBackoffMillis;
        this.jitter = jitter;
    }

    public State state() {
        return state;
    }

    // The timestamp the banner shows. Server clock, because that is the one
    // the price was true at.
    public long lastServerMillis() {
        return lastServerMillis;
    }

    // Any message at all, a quote or a heartbeat, resets both the staleness
    // clock and the backoff ladder.
    public void onMessage(long elapsedMillis, long serverMillis) {
        attempt = 0;
        lastMessageElapsed = elapsedMillis;
        lastServerMillis = serverMillis;
        state = State.LIVE;
    }

    // elapsedMillis is the phone's monotonic clock, and it is used for exactly
    // one thing, how long since we last heard anything. It never stamps a
    // price and it never buckets a candle.
    public void onTimeCheck(long elapsedMillis) {
        if (state == State.LIVE && elapsedMillis - lastMessageElapsed > stalenessMillis) state = State.STALE;
    }

    // Returns how long to wait before the next attempt.
    public long onDropped() {
        attempt++;
        state = State.RECONNECTING;
        return backoffMillis(attempt);
    }

    public void onRetry() {
        state = State.CONNECTING;
    }

    public void onBackground() {
        state = State.CLOSED;
    }

    public void onForeground() {
        attempt = 0;
        state = State.CONNECTING;
    }

    // Exponential, capped, and half of it random. The jitter is not a detail.
    // A feed outage drops every phone in the country at the same instant, and
    // without jitter they all come back on the same second and drop it again.
    public long backoffMillis(int attempt) {
        long window = Math.min(maxBackoffMillis, baseBackoffMillis << Math.min(attempt - 1, 16));
        long half = window / 2;
        return half + jitter.applyAsLong(half + 1);
    }

    // Anything but LIVE means the numbers on screen are not current, and the
    // screen says so.
    public boolean showStaleBanner() {
        return state != State.LIVE;
    }
}

package com.androidinterview.trading.connection;

import java.util.function.LongUnaryOperator;

public final class FeedConnection {

    public enum State {
        CONNECTING, LIVE,
        STALE,
        RECONNECTING,
        CLOSED
    }

    private final long stalenessMillis;
    private final long baseBackoffMillis;
    private final long maxBackoffMillis;
    private final LongUnaryOperator jitter;

    private State state = State.CONNECTING;
    private int attempt;
    private long lastMessageElapsed;
    private long lastServerMillis;

    public FeedConnection(long stalenessMillis, long baseBackoffMillis, long maxBackoffMillis, LongUnaryOperator jitter) {
        this.stalenessMillis = stalenessMillis;
        this.baseBackoffMillis = baseBackoffMillis;
        this.maxBackoffMillis = maxBackoffMillis;
        this.jitter = jitter;
    }

    public State state() {
        return state;
    }

    public long lastServerMillis() {
        return lastServerMillis;
    }

    public void onMessage(long elapsedMillis, long serverMillis) {
        attempt = 0;
        lastMessageElapsed = elapsedMillis;
        lastServerMillis = serverMillis;
        state = State.LIVE;
    }

    public void onTimeCheck(long elapsedMillis) {
        if (state == State.LIVE && elapsedMillis - lastMessageElapsed > stalenessMillis) state = State.STALE;
    }

    public long onDropped() {
        attempt++;
        state = State.RECONNECTING;
        return backoffMillis(attempt);
    }

    public void onRetry() {
        state = State.CONNECTING;
    }

    public void onBackground() {
        state = State.CLOSED;
    }

    public void onForeground() {
        attempt = 0;
        state = State.CONNECTING;
    }

    public long backoffMillis(int attempt) {
        long window = Math.min(maxBackoffMillis, baseBackoffMillis << Math.min(attempt - 1, 16));
        long half = window / 2;
        return half + jitter.applyAsLong(half + 1);
    }

    public boolean showStaleBanner() {
        return state != State.LIVE;
    }
}

package com.androidinterview.trading.feed;

import com.androidinterview.trading.feed.MarketMessage.BookDelta;
import com.androidinterview.trading.feed.MarketMessage.BookSnapshot;
import com.androidinterview.trading.feed.MarketMessage.Trade;
import java.util.HashMap;
import java.util.Map;

// The one piece of the client that decides whether the numbers on screen are
// real. One socket carries every subscribed symbol, and this class routes by
// symbol and refuses anything it cannot prove is in order.
public final class MarketDataFeed {

    // OkHttp's WebSocket sits behind this in a real app. Subscribing to a
    // symbol makes the server send a snapshot first and deltas after it.
    public interface Socket {
        void subscribe(String symbol);

        void unsubscribe(String symbol);
    }

    // Only messages that passed the sequence check reach here.
    public interface Sink {
        void onSnapshot(BookSnapshot snapshot);

        void onDelta(BookDelta delta);

        void onTrade(Trade trade);
    }

    public enum Outcome {
        APPLIED,
        // A replay, or a delta that arrived while we are still waiting for a
        // snapshot. Dropped, silently and on purpose.
        IGNORED,
        // A sequence we never saw. The book is now unknowable, so we throw it
        // away and ask for a fresh one.
        GAP
    }

    private final Socket socket;
    private final Sink sink;

    // The next sequence each symbol owes us. No entry means we are waiting for
    // a snapshot and have nothing to apply deltas to.
    private final Map<String, Long> expected = new HashMap<>();

    public MarketDataFeed(Socket socket, Sink sink) {
        this.socket = socket;
        this.sink = sink;
    }

    // Subscribe only to what is on screen. A watchlist of eight symbols is
    // eight subscriptions, not the whole exchange, and scrolling a symbol off
    // screen unsubscribes it. That is most of the battery and data budget.
    public void subscribe(String symbol) {
        expected.remove(symbol);
        socket.subscribe(symbol);
    }

    public void unsubscribe(String symbol) {
        expected.remove(symbol);
        socket.unsubscribe(symbol);
    }

    public Outcome onMessage(MarketMessage message) {
        if (message instanceof BookSnapshot snapshot) {
            // A snapshot is accepted unconditionally, because it is the base
            // and not a patch. It also resets the ladder after a gap.
            expected.put(snapshot.symbol(), snapshot.seq() + 1);
            sink.onSnapshot(snapshot);
            return Outcome.APPLIED;
        }

        Long want = expected.get(message.symbol());
        if (want == null) return Outcome.IGNORED;        // no base yet
        if (message.seq() < want) return Outcome.IGNORED; // already applied

        if (message.seq() > want) {
            // The gap. Never patch across it. A book missing one delta looks
            // completely normal and is wrong forever, and a wrong book is how
            // someone sells into a bid that is not there.
            resnapshot(message.symbol());
            return Outcome.GAP;
        }

        expected.put(message.symbol(), message.seq() + 1);
        if (message instanceof BookDelta delta) sink.onDelta(delta);
        else if (message instanceof Trade trade) sink.onTrade(trade);
        return Outcome.APPLIED;
    }

    // Drop the sequence, drop the subscription, ask again. Deltas that arrive
    // in the meantime are ignored until the new snapshot lands, which is the
    // point, a few hundred milliseconds of a frozen book beats an hour of a
    // plausible wrong one.
    private void resnapshot(String symbol) {
        expected.remove(symbol);
        socket.unsubscribe(symbol);
        socket.subscribe(symbol);
    }
}

package com.androidinterview.trading.feed;

import com.androidinterview.trading.feed.MarketMessage.BookDelta;
import com.androidinterview.trading.feed.MarketMessage.BookSnapshot;
import com.androidinterview.trading.feed.MarketMessage.Trade;
import java.util.HashMap;
import java.util.Map;

public final class MarketDataFeed {

    public interface Socket {
        void subscribe(String symbol);

        void unsubscribe(String symbol);
    }

    public interface Sink {
        void onSnapshot(BookSnapshot snapshot);

        void onDelta(BookDelta delta);

        void onTrade(Trade trade);
    }

    public enum Outcome {
        APPLIED,
        IGNORED,
        GAP
    }

    private final Socket socket;
    private final Sink sink;

    private final Map<String, Long> expected = new HashMap<>();

    public MarketDataFeed(Socket socket, Sink sink) {
        this.socket = socket;
        this.sink = sink;
    }

    public void subscribe(String symbol) {
        expected.remove(symbol);
        socket.subscribe(symbol);
    }

    public void unsubscribe(String symbol) {
        expected.remove(symbol);
        socket.unsubscribe(symbol);
    }

    public Outcome onMessage(MarketMessage message) {
        if (message instanceof BookSnapshot snapshot) {
            expected.put(snapshot.symbol(), snapshot.seq() + 1);
            sink.onSnapshot(snapshot);
            return Outcome.APPLIED;
        }

        Long want = expected.get(message.symbol());
        if (want == null) return Outcome.IGNORED;
        if (message.seq() < want) return Outcome.IGNORED;

        if (message.seq() > want) {
            resnapshot(message.symbol());
            return Outcome.GAP;
        }

        expected.put(message.symbol(), message.seq() + 1);
        if (message instanceof BookDelta delta) sink.onDelta(delta);
        else if (message instanceof Trade trade) sink.onTrade(trade);
        return Outcome.APPLIED;
    }

    private void resnapshot(String symbol) {
        expected.remove(symbol);
        socket.unsubscribe(symbol);
        socket.subscribe(symbol);
    }
}

package com.androidinterview.trading.feed;

import java.util.List;

// Everything one market data socket carries, as a closed set.
//
// Every message carries a sequence number for its symbol, and that number is
// the whole correctness story. A book you patch with deltas is only correct if
// you applied every delta in order, so the client has to be able to prove it
// did, and to notice the moment it cannot.
//
// Prices are integer ticks, not doubles. A penny that rounds is a penny an
// order is placed at.
public sealed interface MarketMessage {

    String symbol();

    long seq();

    // A quantity of zero is a removal, not a level with nothing on it.
    record Level(long priceTicks, long quantity) {
    }

    // The base the deltas are applied to. Sent on subscribe, and again on
    // every resubscribe after a gap.
    record BookSnapshot(String symbol, long seq, List<Level> bids, List<Level> asks) implements MarketMessage {
    }

    record BookDelta(String symbol, long seq, boolean bid, Level level) implements MarketMessage {
    }

    // One executed trade. serverMillis is the exchange's clock, and it is what
    // buckets this trade into a candle. The phone's clock never touches it.
    record Trade(String symbol, long seq, long priceTicks, long quantity, long serverMillis) implements MarketMessage {
    }
}

package com.androidinterview.trading.feed;

import java.util.List;

public sealed interface MarketMessage {

    String symbol();

    long seq();

    record Level(long priceTicks, long quantity) {
    }

    record BookSnapshot(String symbol, long seq, List<Level> bids, List<Level> asks) implements MarketMessage {
    }

    record BookDelta(String symbol, long seq, boolean bid, Level level) implements MarketMessage {
    }

    record Trade(String symbol, long seq, long priceTicks, long quantity, long serverMillis) implements MarketMessage {
    }
}

package com.androidinterview.trading.book

import com.androidinterview.trading.feed.Level
import com.androidinterview.trading.feed.MarketMessage
import com.androidinterview.trading.feed.Price
import com.androidinterview.trading.feed.Quantity
import java.util.TreeMap

// Depth for one symbol, as two sorted maps. Bids descend so the best bid is
// first, asks ascend so the best ask is first, and the top of each side is a
// first key rather than a scan.
//
// Deliberately dumb about ordering. It applies what it is given, and the feed
// above it is what guarantees nothing was skipped.
class OrderBook {

    private val bids = TreeMap<Price, Quantity>(reverseOrder())
    private val asks = TreeMap<Price, Quantity>()

    val bestBid: Price? get() = bids.keys.firstOrNull()
    val bestAsk: Price? get() = asks.keys.firstOrNull()

    // A snapshot replaces the book rather than merging into it, because the
    // whole reason a snapshot arrived is that the old contents are suspect.
    fun reset(snapshot: MarketMessage.BookSnapshot) {
        bids.clear()
        asks.clear()
        snapshot.bids.forEach { apply(bid = true, level = it) }
        snapshot.asks.forEach { apply(bid = false, level = it) }
    }

    // Zero quantity means the level is gone, so it is removed rather than
    // stored. Keeping empty levels leaves rows of nothing at the top of the
    // ladder and quietly breaks the best bid and the best ask.
    fun apply(bid: Boolean, level: Level) {
        val side = if (bid) bids else asks
        if (level.quantity.isEmpty) side.remove(level.price) else side[level.price] = level.quantity
    }

    // The screen shows ten rows a side, not the four thousand levels the feed
    // knows about, so the read is bounded even when the book is not.
    fun top(bid: Boolean, depth: Int): List<Level> =
        (if (bid) bids else asks).entries.take(depth).map { Level(it.key, it.value) }
}

package com.androidinterview.trading.book

import com.androidinterview.trading.feed.Level
import com.androidinterview.trading.feed.MarketMessage
import com.androidinterview.trading.feed.Price
import com.androidinterview.trading.feed.Quantity
import java.util.TreeMap

class OrderBook {

    private val bids = TreeMap<Price, Quantity>(reverseOrder())
    private val asks = TreeMap<Price, Quantity>()

    val bestBid: Price? get() = bids.keys.firstOrNull()
    val bestAsk: Price? get() = asks.keys.firstOrNull()

    fun reset(snapshot: MarketMessage.BookSnapshot) {
        bids.clear()
        asks.clear()
        snapshot.bids.forEach { apply(bid = true, level = it) }
        snapshot.asks.forEach { apply(bid = false, level = it) }
    }

    fun apply(bid: Boolean, level: Level) {
        val side = if (bid) bids else asks
        if (level.quantity.isEmpty) side.remove(level.price) else side[level.price] = level.quantity
    }

    fun top(bid: Boolean, depth: Int): List<Level> =
        (if (bid) bids else asks).entries.take(depth).map { Level(it.key, it.value) }
}

package com.androidinterview.trading.chart

import com.androidinterview.trading.feed.MarketMessage
import com.androidinterview.trading.feed.Price
import com.androidinterview.trading.feed.Quantity

data class Candle(
    val openMillis: Long,
    val open: Price,
    val high: Price,
    val low: Price,
    val close: Price,
    val volume: Quantity,
) {
    fun fold(price: Price, quantity: Quantity) = copy(
        high = maxOf(high, price),
        low = minOf(low, price),
        close = price,
        volume = volume + quantity,
    )
}

// Where a thousand ticks a second becomes sixty frames a second.
//
// The socket thread folds every trade into the live candle and writes the
// result into one field. The UI thread reads that field once a frame and draws
// it. Nothing is queued, nothing is posted, and no tick is ever handed to the
// main thread, so a fast feed costs a little arithmetic on a background thread
// rather than a dropped frame. In a real app `live` is the backing value of a
// StateFlow, which conflates for the same reason and by the same rule.
//
// Historical candles come from REST, once, and are immutable. Only the newest
// candle is built here, because only the newest one is still changing.
class CandleBuilder(private val intervalMillis: Long) {

    init {
        require(intervalMillis > 0) { "interval must be positive" }
    }

    // Volatile, so the UI thread reads a fully published object rather than a
    // half written one. A Candle is immutable, so a sample is always coherent
    // even when the next tick lands mid draw.
    @Volatile
    var live: Candle? = null
        private set

    // A diagnostic, so a test can show how many ticks one sample stood for.
    var foldedTicks = 0L
        private set

    // Called on the socket thread, thousands of times a second. Returns the
    // candle that just closed, so the caller can append it to the history, or
    // null when the live candle simply grew.
    fun fold(trade: MarketMessage.Trade): Candle? {
        foldedTicks++
        // The bucket comes from the exchange's clock. A phone five minutes
        // fast would file this trade in the wrong minute and draw a chart that
        // disagrees with every other screen in the world.
        val bucket = trade.serverMillis - Math.floorMod(trade.serverMillis, intervalMillis)
        val current = live

        return when {
            current == null || bucket > current.openMillis -> current.also {
                live = Candle(bucket, trade.price, trade.price, trade.price, trade.price, trade.quantity)
            }
            // A trade stamped before the live candle opened arrived late. It
            // belongs to a candle already sent to history, so it is dropped
            // rather than allowed to rewrite the past.
            bucket < current.openMillis -> null
            else -> null.also { live = current.fold(trade.price, trade.quantity) }
        }
    }

    // Called on the UI thread, at most once a frame. Every tick between two
    // calls is already folded into what this returns, which is the whole
    // point, the intermediate states were never worth drawing.
    fun sample(): Candle? = live
}

package com.androidinterview.trading.chart

import com.androidinterview.trading.feed.MarketMessage
import com.androidinterview.trading.feed.Price
import com.androidinterview.trading.feed.Quantity

data class Candle(
    val openMillis: Long,
    val open: Price,
    val high: Price,
    val low: Price,
    val close: Price,
    val volume: Quantity,
) {
    fun fold(price: Price, quantity: Quantity) = copy(
        high = maxOf(high, price),
        low = minOf(low, price),
        close = price,
        volume = volume + quantity,
    )
}

class CandleBuilder(private val intervalMillis: Long) {

    init {
        require(intervalMillis > 0) { "interval must be positive" }
    }

    @Volatile
    var live: Candle? = null
        private set

    var foldedTicks = 0L
        private set

    fun fold(trade: MarketMessage.Trade): Candle? {
        foldedTicks++
        val bucket = trade.serverMillis - Math.floorMod(trade.serverMillis, intervalMillis)
        val current = live

        return when {
            current == null || bucket > current.openMillis -> current.also {
                live = Candle(bucket, trade.price, trade.price, trade.price, trade.price, trade.quantity)
            }
            bucket < current.openMillis -> null
            else -> null.also { live = current.fold(trade.price, trade.quantity) }
        }
    }

    fun sample(): Candle? = live
}

package com.androidinterview.trading.connection

import kotlin.math.min
import kotlin.random.Random

// The lifecycle of the market data socket, as five states the UI can render
// honestly. The failure this exists to prevent is the silent one. A socket
// that stops delivering looks exactly like a market that stopped moving, and a
// trader who cannot tell those apart will place an order against a price from
// four minutes ago.
sealed interface ConnectionState {
    data object Connecting : ConnectionState
    data object Live : ConnectionState

    // Still connected, but nothing has arrived for longer than it should have.
    // The prices stay on screen, greyed, with the server timestamp they were
    // true at, rather than pretending to be current.
    data class Stale(val lastServerMillis: Long) : ConnectionState

    data class Reconnecting(val attempt: Int, val retryInMillis: Long) : ConnectionState

    // Backgrounded. Holding a market data socket behind a locked screen burns
    // battery to render nothing, so it closes and a resume takes a fresh
    // snapshot.
    data object Closed : ConnectionState
}

class FeedConnection(
    private val stalenessMillis: Long = 5_000,
    private val baseBackoffMillis: Long = 500,
    private val maxBackoffMillis: Long = 30_000,
    // Injected so a test is deterministic.
    private val jitter: (Long) -> Long = { Random.nextLong(it) },
) {
    var state: ConnectionState = ConnectionState.Connecting
        private set

    private var attempt = 0
    private var lastMessageElapsed = 0L
    private var lastServerMillis = 0L

    // Any message at all, a quote or a heartbeat, resets both the staleness
    // clock and the backoff ladder.
    fun onMessage(elapsedMillis: Long, serverMillis: Long) {
        attempt = 0
        lastMessageElapsed = elapsedMillis
        lastServerMillis = serverMillis
        state = ConnectionState.Live
    }

    // elapsedMillis is the phone's monotonic clock, used for exactly one
    // thing, how long since we last heard anything. It never stamps a price
    // and it never buckets a candle.
    fun onTimeCheck(elapsedMillis: Long) {
        if (state == ConnectionState.Live && elapsedMillis - lastMessageElapsed > stalenessMillis) {
            state = ConnectionState.Stale(lastServerMillis)
        }
    }

    fun onDropped(): Long {
        attempt++
        return backoffMillis(attempt).also { state = ConnectionState.Reconnecting(attempt, it) }
    }

    fun onRetry() {
        state = ConnectionState.Connecting
    }

    fun onBackground() {
        state = ConnectionState.Closed
    }

    fun onForeground() {
        attempt = 0
        state = ConnectionState.Connecting
    }

    // Exponential, capped, and half of it random. The jitter is not a detail.
    // A feed outage drops every phone in the country at the same instant, and
    // without it they all come back on the same second and drop it again.
    fun backoffMillis(attempt: Int): Long {
        val window = min(maxBackoffMillis, baseBackoffMillis shl min(attempt - 1, 16))
        val half = window / 2
        return half + jitter(half + 1)
    }

    // Anything but Live means the numbers on screen are not current, and the
    // screen says so.
    val stale: Boolean get() = state != ConnectionState.Live
}

package com.androidinterview.trading.connection

import kotlin.math.min
import kotlin.random.Random

sealed interface ConnectionState {
    data object Connecting : ConnectionState
    data object Live : ConnectionState

    data class Stale(val lastServerMillis: Long) : ConnectionState

    data class Reconnecting(val attempt: Int, val retryInMillis: Long) : ConnectionState

    data object Closed : ConnectionState
}

class FeedConnection(
    private val stalenessMillis: Long = 5_000,
    private val baseBackoffMillis: Long = 500,
    private val maxBackoffMillis: Long = 30_000,
    private val jitter: (Long) -> Long = { Random.nextLong(it) },
) {
    var state: ConnectionState = ConnectionState.Connecting
        private set

    private var attempt = 0
    private var lastMessageElapsed = 0L
    private var lastServerMillis = 0L

    fun onMessage(elapsedMillis: Long, serverMillis: Long) {
        attempt = 0
        lastMessageElapsed = elapsedMillis
        lastServerMillis = serverMillis
        state = ConnectionState.Live
    }

    fun onTimeCheck(elapsedMillis: Long) {
        if (state == ConnectionState.Live && elapsedMillis - lastMessageElapsed > stalenessMillis) {
            state = ConnectionState.Stale(lastServerMillis)
        }
    }

    fun onDropped(): Long {
        attempt++
        return backoffMillis(attempt).also { state = ConnectionState.Reconnecting(attempt, it) }
    }

    fun onRetry() {
        state = ConnectionState.Connecting
    }

    fun onBackground() {
        state = ConnectionState.Closed
    }

    fun onForeground() {
        attempt = 0
        state = ConnectionState.Connecting
    }

    fun backoffMillis(attempt: Int): Long {
        val window = min(maxBackoffMillis, baseBackoffMillis shl min(attempt - 1, 16))
        val half = window / 2
        return half + jitter(half + 1)
    }

    val stale: Boolean get() = state != ConnectionState.Live
}

package com.androidinterview.trading.feed

// Prices are integer ticks and quantities are integer units, both wrapped so
// the compiler stops you adding a price to a size. A double price is a penny
// that rounds, and a penny that rounds is a penny an order is placed at.
@JvmInline
value class Price(val ticks: Long) : Comparable<Price> {
    override fun compareTo(other: Price) = ticks.compareTo(other.ticks)
}

@JvmInline
value class Quantity(val units: Long) {
    val isEmpty get() = units == 0L
    operator fun plus(other: Quantity) = Quantity(units + other.units)
}

// A quantity of zero is a removal, not a level with nothing on it.
data class Level(val price: Price, val quantity: Quantity)

// Everything one market data socket carries, as a closed set. Every message
// carries a sequence number for its symbol, and that number is the whole
// correctness story. A book patched with deltas is only correct if every delta
// was applied in order, so the client has to be able to prove it was and to
// notice the moment it cannot.
sealed interface MarketMessage {
    val symbol: String
    val seq: Long

    // The base the deltas are applied to. Sent on subscribe, and again on
    // every resubscribe after a gap.
    data class BookSnapshot(
        override val symbol: String,
        override val seq: Long,
        val bids: List<Level>,
        val asks: List<Level>,
    ) : MarketMessage

    data class BookDelta(
        override val symbol: String,
        override val seq: Long,
        val bid: Boolean,
        val level: Level,
    ) : MarketMessage

    // One executed trade. serverMillis is the exchange's clock, and it is what
    // buckets this trade into a candle. The phone's clock never touches it.
    data class Trade(
        override val symbol: String,
        override val seq: Long,
        val price: Price,
        val quantity: Quantity,
        val serverMillis: Long,
    ) : MarketMessage
}

// OkHttp's WebSocket sits behind this. Subscribing makes the server send a
// snapshot first and deltas after it.
interface MarketSocket {
    fun subscribe(symbol: String)
    fun unsubscribe(symbol: String)
}

enum class Outcome {
    APPLIED,

    // A replay, or a delta that arrived while we are still waiting for a
    // snapshot. Dropped, silently and on purpose.
    IGNORED,

    // A sequence we never saw. The book is now unknowable, so it is thrown
    // away and a fresh one is requested.
    GAP,
}

// The one piece of the client that decides whether the numbers on screen are
// real. One socket carries every subscribed symbol, this routes by symbol and
// refuses anything it cannot prove is in order.
class MarketDataFeed(
    private val socket: MarketSocket,
    private val onApplied: (MarketMessage) -> Unit,
) {
    // The next sequence each symbol owes us. No entry means we are waiting for
    // a snapshot and have nothing to apply deltas to.
    private val expected = mutableMapOf<String, Long>()

    // Subscribe only to what is on screen. A watchlist of eight symbols is
    // eight subscriptions, not the whole exchange, and scrolling a symbol off
    // screen unsubscribes it. That is most of the battery and data budget.
    fun subscribe(symbol: String) {
        expected -= symbol
        socket.subscribe(symbol)
    }

    fun unsubscribe(symbol: String) {
        expected -= symbol
        socket.unsubscribe(symbol)
    }

    fun onMessage(message: MarketMessage): Outcome {
        // A snapshot is accepted unconditionally, because it is the base and
        // not a patch. It also resets the ladder after a gap.
        if (message is MarketMessage.BookSnapshot) {
            expected[message.symbol] = message.seq + 1
            onApplied(message)
            return Outcome.APPLIED
        }
        val want = expected[message.symbol] ?: return Outcome.IGNORED
        return when {
            message.seq < want -> Outcome.IGNORED
            // The gap. Never patch across it. A book missing one delta looks
            // completely normal and is wrong forever, and a wrong book is how
            // someone sells into a bid that is not there.
            message.seq > want -> resnapshot(message.symbol)
            else -> {
                expected[message.symbol] = message.seq + 1
                onApplied(message)
                Outcome.APPLIED
            }
        }
    }

    // Drop the sequence, drop the subscription, ask again. Deltas arriving in
    // the meantime are ignored until the new snapshot lands, which is the
    // point. A few hundred milliseconds of a frozen book beats an hour of a
    // plausible wrong one.
    private fun resnapshot(symbol: String): Outcome {
        expected -= symbol
        socket.unsubscribe(symbol)
        socket.subscribe(symbol)
        return Outcome.GAP
    }
}

package com.androidinterview.trading.feed

@JvmInline
value class Price(val ticks: Long) : Comparable<Price> {
    override fun compareTo(other: Price) = ticks.compareTo(other.ticks)
}

@JvmInline
value class Quantity(val units: Long) {
    val isEmpty get() = units == 0L
    operator fun plus(other: Quantity) = Quantity(units + other.units)
}

data class Level(val price: Price, val quantity: Quantity)

sealed interface MarketMessage {
    val symbol: String
    val seq: Long

    data class BookSnapshot(
        override val symbol: String,
        override val seq: Long,
        val bids: List<Level>,
        val asks: List<Level>,
    ) : MarketMessage

    data class BookDelta(
        override val symbol: String,
        override val seq: Long,
        val bid: Boolean,
        val level: Level,
    ) : MarketMessage

    data class Trade(
        override val symbol: String,
        override val seq: Long,
        val price: Price,
        val quantity: Quantity,
        val serverMillis: Long,
    ) : MarketMessage
}

interface MarketSocket {
    fun subscribe(symbol: String)
    fun unsubscribe(symbol: String)
}

enum class Outcome {
    APPLIED,

    IGNORED,

    GAP,
}

class MarketDataFeed(
    private val socket: MarketSocket,
    private val onApplied: (MarketMessage) -> Unit,
) {
    private val expected = mutableMapOf<String, Long>()

    fun subscribe(symbol: String) {
        expected -= symbol
        socket.subscribe(symbol)
    }

    fun unsubscribe(symbol: String) {
        expected -= symbol
        socket.unsubscribe(symbol)
    }

    fun onMessage(message: MarketMessage): Outcome {
        if (message is MarketMessage.BookSnapshot) {
            expected[message.symbol] = message.seq + 1
            onApplied(message)
            return Outcome.APPLIED
        }
        val want = expected[message.symbol] ?: return Outcome.IGNORED
        return when {
            message.seq < want -> Outcome.IGNORED
            message.seq > want -> resnapshot(message.symbol)
            else -> {
                expected[message.symbol] = message.seq + 1
                onApplied(message)
                Outcome.APPLIED
            }
        }
    }

    private fun resnapshot(symbol: String): Outcome {
        expected -= symbol
        socket.unsubscribe(symbol)
        socket.subscribe(symbol)
        return Outcome.GAP
    }
}