Exercise: Market data processing optimization
Your task is to optimize the code used to observe market data, and to display events that fulfill the given filters. The code is already working, but it is not optimized. The Filter class is used to filter events. The MarketRepository class is used to observe market data and to emit events. The MarketRepository class is used to emit market events. TradeService is used to observe market data and to display events that fulfill the given filters. Other classes are used to model market data, and to provide fake data for tests. Check how much time this code takes before and after each optimization.
import effective.efficient.Filter.*
import effective.efficient.Filter.Relation.*
import effective.efficient.Filter.SnapshotPart.*
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.launch
import java.util.concurrent.ConcurrentHashMap
import kotlin.random.Random
import kotlin.system.measureTimeMillis
data class TickerSnapshot(
val ticker: Ticker,
val snapshot: Snapshot,
)
data class Snapshot(
val bid: PriceSizeTime?,
val ask: PriceSizeTime?,
val last: PriceSizeTime?,
)
data class PriceSizeTime(
val price: Price,
val size: Int? = null,
val time: Long? = null,
)
data class Ticker(val value: String)
data class Price(val value: Float?)
sealed interface Event { val ticker: String }
data class BidEvent(override val ticker: String, val price: Float?, val size: Int?, val time: Long?) : Event
data class AskEvent(override val ticker: String, val price: Float?, val size: Int?, val time: Long?) : Event
data class TradeEvent(override val ticker: String, val price: Float?, val size: Int?, val time: Long?) : Event
val tickers = List(1000) { Ticker("Ticker$it") }
// Do not touch this one
class MarketClient {
fun observe() = flow {
val random = Random(123456789)
while (true) {
val event = when ((0..2).random(random)) {
0 -> BidEvent(
tickers.random(random).value,
if (random.nextInt(100) == 1) null else (0..100).random(random).toFloat(),
if (random.nextInt(100) == 1) null else (0..100).random(random),
if (random.nextInt(100) == 1) null else System.currentTimeMillis()
)
1 -> AskEvent(
tickers.random(random).value,
if (random.nextInt(100) == 1) null else (0..100).random(random).toFloat(),
if (random.nextInt(100) == 1) null else (0..100).random(random),
if (random.nextInt(100) == 1) null else System.currentTimeMillis()
)
else -> TradeEvent(
tickers.random(random).value,
if (random.nextInt(100) == 1) null else (0..100).random(random).toFloat(),
if (random.nextInt(100) == 1) null else (0..100).random(random),
if (random.nextInt(100) == 1) null else System.currentTimeMillis()
)
}
emit(event)
}
}
}
class MarketRepository(
private val client: MarketClient,
backgroundScope: CoroutineScope,
) {
private val snapshots = ConcurrentHashMap<Ticker, Snapshot>()
private val updates = MutableSharedFlow<TickerSnapshot>()
fun observeUpdates() = updates
.onStart { snapshots.forEach { emit(TickerSnapshot(it.key, it.value)) } }
init {
backgroundScope.launch {
client.observe().collect {
when (it) {
is BidEvent -> {
val snapshot = snapshots.getOrPut(Ticker(it.ticker)) { Snapshot(null, null, null) }
.copy(bid = PriceSizeTime(Price(it.price), it.size, it.time))
snapshots[Ticker(it.ticker)] = snapshot
updates.emit(TickerSnapshot(Ticker(it.ticker), snapshot))
}
is AskEvent -> {
val snapshot = snapshots.getOrPut(Ticker(it.ticker)) { Snapshot(null, null, null) }
.copy(ask = PriceSizeTime(Price(it.price), it.size, it.time))
snapshots[Ticker(it.ticker)] = snapshot
updates.emit(TickerSnapshot(Ticker(it.ticker), snapshot))
}
is TradeEvent -> {
val snapshot = snapshots.getOrPut(Ticker(it.ticker)) { Snapshot(null, null, null) }
.copy(last = PriceSizeTime(Price(it.price), it.size, it.time))
snapshots[Ticker(it.ticker)] = snapshot
updates.emit(TickerSnapshot(Ticker(it.ticker), snapshot))
}
}
}
}
}
}
sealed class Filter {
data object All : Filter()
class Or(val filters: List<Filter>) : Filter()
class And(val filters: List<Filter>) : Filter()
class PrizeCondition(
val snapshotPart: SnapshotPart,
val relation: Relation,
val value: Float,
) : Filter()
class TickerIs(val tickers: List<Ticker>) : Filter()
class Not(val filter: Filter) : Filter()
enum class SnapshotPart {
Ask, Bid, Last, Spread
}
enum class Relation {
GreaterThan, LessThan, Equal
}
fun check(tickerSnapshot: TickerSnapshot): Boolean = when (this) {
All -> true
is Or -> filters.any { it.check(tickerSnapshot) }
is And -> filters.all { it.check(tickerSnapshot) }
is PrizeCondition -> run {
val snapshotPrize = when (snapshotPart) {
Ask -> tickerSnapshot.snapshot.ask?.price?.value ?: return@run false
Bid -> tickerSnapshot.snapshot.bid?.price?.value ?: return@run false
Last -> tickerSnapshot.snapshot.last?.price?.value ?: return@run false
Spread -> {
val bid = tickerSnapshot.snapshot.bid?.price?.value ?: return@run false
val ask = tickerSnapshot.snapshot.ask?.price?.value ?: return@run false
ask - bid
}
}
when (relation) {
GreaterThan -> snapshotPrize > value
LessThan -> snapshotPrize < value
Equal -> snapshotPrize == value
}
}
is TickerIs -> tickers.contains(tickerSnapshot.ticker)
is Not -> !filter.check(tickerSnapshot)
}
}
class TradeService(
private val repository: MarketRepository,
) {
fun observeUpdates(
filter: Filter,
tickers: List<Ticker>? = null,
) = repository.observeUpdates()
.filter { tickers == null || it.ticker in tickers }
.filter { filter.check(it) }
}
suspend fun main() {
val client = MarketClient()
val repository = MarketRepository(client, backgroundScope = CoroutineScope(SupervisorJob()))
val service = TradeService(repository)
val filter = Or(
listOf(
And(listOf(TickerIs(tickers.take(1)), PrizeCondition(Ask, GreaterThan, 99f))),
And(listOf(PrizeCondition(Spread, GreaterThan, 99f))),
)
)
measureTimeMillis {
service.observeUpdates(
filter = filter,
tickers = tickers.take(70)
).take(1_000)
.collect { println(it) }
}.let { println("Took $it") }
}
Once you are done with the exercise, you can check your solution here.
