Merge pull request #208 from cybercongress/206-tickers-endpoint

Tickers rest endpoint

build.gradle

@@ -27,10 +27,10 @@ buildscript {
  shadowPluginVersion = "2.0.1"
 
  // tests
- junitVersion = "5.0.3"
+ junitVersion = "5.2.0"
  junitPlatformVersion = "1.0.3"
- mockitoVersion = "2.1.0"
- mockitoKotlinVersion = "0.7.0"
+ mockitoVersion = "2.18.3"
+ mockitoKotlinVersion = "1.6.0"
 
  // logs
  slf4jVersion = "1.7.25"
@@ -235,6 +235,9 @@ project(":rest-api") {
  compile project(":common")
  compile project(":common-rest-api")
  compile project(":cassandra-service")
+
+ testCompile 'org.springframework.boot:spring-boot-starter-test'
+ testCompile 'com.nhaarman:mockito-kotlin'
  }
 }
 

cassandra-service/src/main/kotlin/fund/cyber/markets/cassandra/repository/TickerRepository.kt

@@ -24,4 +24,15 @@ interface TickerRepository : ReactiveCassandraRepository<CqlTokenTicker, MapId>
  @Param("timestampFrom") timestampFrom: Date,
  @Param("timestampTo") timestampTo: Date,
  @Param("interval") interval: Long): Flux<CqlTokenTicker>
+
+ @Consistency(value = ConsistencyLevel.LOCAL_QUORUM)
+ @Query("""
+ SELECT * FROM markets.ticker
+ WHERE tokenSymbol=:tokenSymbol AND epochDay=:epochDay AND interval=:interval
+ AND timestampFrom<=:timestampFrom LIMIT :limitValue""")
+ fun find(@Param("tokenSymbol") tokenSymbol: String,
+ @Param("epochDay") epochDay: Long,
+ @Param("timestampFrom") timestampFrom: Date,
+ @Param("interval") interval: Long,
+ @Param("limitValue") limit: Long): Flux<CqlTokenTicker>
 }

common/src/main/kotlin/fund/cyber/markets/common/Converter.kt

@@ -13,6 +13,8 @@ const val MINUTES_TO_MILLIS: Double = 1000.0 * 60
 const val MINUTES_TO_SECONDS: Double = 60.0
 const val MINUTES_TO_HOURS: Double = 1.0 / 60
 
+const val DAYS_TO_MILLIS: Double = 1000.0 * 60 * 60 *24
+
 infix fun Long.convert(coefficient: Double): Long {
  return (this * coefficient).toLong()
 }

common/src/main/kotlin/fund/cyber/markets/common/Env.kt

@@ -34,4 +34,7 @@ const val SAVE_ORDERBOOKS = "SAVE_ORDERBOOKS"
 const val SAVE_ORDERBOOKS_DEFAULT = true
 
 const val LAG_FROM_REAL_TIME_MIN = "LAG_FROM_REAL_TIME_MIN"
-const val LAG_FROM_REAL_TIME_MIN_DEFAULT: Long = 60 * 3
+const val LAG_FROM_REAL_TIME_MIN_DEFAULT: Long = 60 * 3
+
+const val CORS_ALLOWED_ORIGINS = "CORS_ALLOWED_ORIGINS"
+const val CORS_ALLOWED_ORIGINS_DEFAULT = "cybermarkets.sh"

dev-environment/elassandra-bootstrap.cql

@@ -32,6 +32,9 @@ CREATE TABLE IF NOT EXISTS markets.ticker (
  price blob,
 
  PRIMARY KEY ((tokenSymbol, epochDay, interval), timestampFrom)
+)
+WITH CLUSTERING ORDER BY (
+ timestampFrom DESC
 );
 
 CREATE TABLE IF NOT EXISTS markets.trade_last_timestamp (

rest-api/src/main/kotlin/fund/cyber/markets/api/rest/configuration/RestApiConfiguration.kt

@@ -1,23 +1,43 @@
 package fund.cyber.markets.api.rest.configuration
 
+import fund.cyber.markets.common.CORS_ALLOWED_ORIGINS
+import fund.cyber.markets.common.CORS_ALLOWED_ORIGINS_DEFAULT
 import fund.cyber.markets.common.EXCHANGES_CONNECTOR_API_URLS
 import fund.cyber.markets.common.EXCHANGES_CONNECTOR_API_URLS_DEFAULT
 import org.springframework.beans.factory.annotation.Value
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
+import org.springframework.web.cors.CorsConfiguration
+import org.springframework.web.cors.reactive.CorsWebFilter
+import org.springframework.web.cors.reactive.UrlBasedCorsConfigurationSource
+import org.springframework.web.util.pattern.PathPatternParser
 
-const val PAGE_SIZE_DEFAULT = 10
-const val PAGE_DEFAULT = 0
 
 @Configuration
 class RestApiConfiguration(
  @Value("\${$EXCHANGES_CONNECTOR_API_URLS:$EXCHANGES_CONNECTOR_API_URLS_DEFAULT}")
- val exchangesConnectorApiUrls: String
+ private val exchangesConnectorApiUrls: String,
+
+ @Value("\${$CORS_ALLOWED_ORIGINS:$CORS_ALLOWED_ORIGINS_DEFAULT}")
+ private val allowedOrigin: String
 ) {
 
  @Bean
  fun connectorApiUrls(): List<String> {
  return exchangesConnectorApiUrls.split(",").map { url -> url.trim() }
  }
 
+ @Bean
+ fun corsFilter(): CorsWebFilter {
+
+ val config = CorsConfiguration()
+ config.addAllowedOrigin(allowedOrigin)
+ config.addAllowedHeader("*")
+ config.addAllowedMethod("*")
+
+ val source = UrlBasedCorsConfigurationSource(PathPatternParser())
+ source.registerCorsConfiguration("/**", config)
+
+ return CorsWebFilter(source)
+ }
 }

rest-api/src/main/kotlin/fund/cyber/markets/api/rest/handler/TickerHandler.kt

@@ -1,47 +1,38 @@
 package fund.cyber.markets.api.rest.handler
 
-import fund.cyber.markets.api.rest.configuration.PAGE_DEFAULT
-import fund.cyber.markets.api.rest.configuration.PAGE_SIZE_DEFAULT
-import fund.cyber.markets.cassandra.model.CqlTokenTicker
-import fund.cyber.markets.cassandra.repository.TickerRepository
-import fund.cyber.markets.common.model.Exchanges
+import fund.cyber.markets.api.rest.service.TickerService
+import fund.cyber.markets.common.MINUTES_TO_MILLIS
+import fund.cyber.markets.common.convert
+import fund.cyber.markets.common.rest.asServerResponse
 import org.springframework.http.HttpStatus
 import org.springframework.stereotype.Component
 import org.springframework.web.reactive.function.server.ServerRequest
 import org.springframework.web.reactive.function.server.ServerResponse
-import org.springframework.web.reactive.function.server.ServerResponse.notFound
-import org.springframework.web.reactive.function.server.ServerResponse.ok
 import reactor.core.publisher.Mono
 import java.util.*
 
+const val LIMIT_DEFAULT = "100"
+
 @Component
 class TickerHandler(
- private val tickerRepository: TickerRepository
+ private val tickerService: TickerService
 ) {
 
  fun getTickers(serverRequest: ServerRequest): Mono<ServerResponse> {
  val symbol: String
  val ts: Long
+ val interval: Long
 
+ val limit = serverRequest.queryParam("limit").orElse(LIMIT_DEFAULT).toLong()
  try {
  symbol = serverRequest.queryParam("symbol").get().toUpperCase()
  ts = serverRequest.queryParam("ts").get().toLong()
+ interval = serverRequest.queryParam("interval").get().toLong() convert MINUTES_TO_MILLIS
  } catch (e: NoSuchElementException) {
  return ServerResponse.status(HttpStatus.BAD_REQUEST).build()
  }
 
- val exchange = serverRequest.queryParam("exchange").orElse(Exchanges.ALL).toUpperCase()
- val page = serverRequest.queryParam("page").orElse(PAGE_DEFAULT.toString()).toLong()
- val pageSize = serverRequest.queryParam("pageSize").orElse(PAGE_SIZE_DEFAULT.toString()).toLong()
-
- //todo: use correct repository call
- val tickers = tickerRepository.findAll()
-
- return ok()
- .body(tickers, CqlTokenTicker::class.java)
- .switchIfEmpty(
- notFound().build()
- )
+ return tickerService.getTickers(symbol, ts, interval, limit).asServerResponse()
  }
 
 }

rest-api/src/main/kotlin/fund/cyber/markets/api/rest/service/TickerService.kt

@@ -0,0 +1,47 @@
+package fund.cyber.markets.api.rest.service
+
+import fund.cyber.markets.cassandra.model.CqlTokenTicker
+import fund.cyber.markets.cassandra.repository.TickerRepository
+import fund.cyber.markets.common.DAYS_TO_MILLIS
+import fund.cyber.markets.common.MILLIS_TO_DAYS
+import fund.cyber.markets.common.convert
+import org.springframework.stereotype.Service
+import reactor.core.publisher.Flux
+import java.util.*
+
+@Service
+class TickerService(
+ private val tickerRepository: TickerRepository
+) {
+
+ fun getTickers(symbol: String, ts: Long, interval: Long, limit: Long): Flux<CqlTokenTicker> {
+ var tickers = Flux.empty<CqlTokenTicker>()
+
+ var tsVar = ts
+ var limitVar = limit
+
+ while (limitVar > 0) {
+
+ val epochDay = tsVar convert MILLIS_TO_DAYS
+ var iterationLimit = (((epochDay + 1) convert DAYS_TO_MILLIS) - tsVar) / interval
+
+ if (iterationLimit > limitVar) {
+ iterationLimit = limitVar
+ }
+
+ if (iterationLimit > 0) {
+ tickers = tickers
+ .concatWith(
+ tickerRepository.find(symbol, epochDay, Date(ts), interval, iterationLimit)
+ )
+ } else {
+ tsVar = (epochDay + 1) convert DAYS_TO_MILLIS
+ }
+
+ tsVar += interval * iterationLimit
+ limitVar -= iterationLimit
+ }
+
+ return tickers
+ }
+}

rest-api/src/test/kotlin/fund/cyber/markets/api/rest/TickerServiceTest.kt

@@ -0,0 +1,143 @@
+package fund.cyber.markets.api.rest
+
+import com.nhaarman.mockito_kotlin.doReturn
+import com.nhaarman.mockito_kotlin.mock
+import fund.cyber.markets.api.rest.service.TickerService
+import fund.cyber.markets.cassandra.model.CqlTokenTicker
+import fund.cyber.markets.cassandra.repository.TickerRepository
+import fund.cyber.markets.common.Intervals
+import fund.cyber.markets.common.model.TokenTicker
+import org.assertj.core.api.Assertions
+import org.junit.Before
+import org.junit.Test
+import reactor.core.publisher.Flux
+import java.util.*
+
+
+class TickerServiceTest {
+
+ private lateinit var tickerService: TickerService
+
+ @Before
+ fun before() {
+ val repository: TickerRepository = mock {
+ //1m tickers
+ on {
+ find("BTC", 0, Date(0), Intervals.MINUTE, Intervals.DAY / Intervals.MINUTE)
+ }.doReturn(Flux.fromIterable(
+ generateTestData(0L, Intervals.MINUTE, Intervals.DAY / Intervals.MINUTE)
+ ))
+
+ //24h tickers
+ on {
+ find("BTC", 0, Date(0), Intervals.DAY, 1)
+ }.doReturn(Flux.fromIterable(
+ generateTestData(0L, Intervals.DAY, 1)
+ ))
+ on {
+ find("BTC", 1, Date(0), Intervals.DAY, 1)
+ }.doReturn(Flux.fromIterable(
+ generateTestData(Intervals.DAY, Intervals.DAY, 1)
+ ))
+ on {
+ find("BTC", 2, Date(0), Intervals.DAY, 1)
+ }.doReturn(Flux.fromIterable(
+ generateTestData(Intervals.DAY * 2, Intervals.DAY, 1)
+ ))
+
+ //15m tickers
+ on {
+ find("BTC", 0, Date(94 * 15 * 60 * 1000), 15 * 60 * 1000, 2)
+ }.doReturn(Flux.fromIterable(
+ generateTestData(94 * 15 * 60 * 1000, 15 * 60 * 1000, 2)
+ ))
+ on {
+ find("BTC", 1, Date(94 * 15 * 60 * 1000), 15 * 60 * 1000, 96)
+ }.doReturn(Flux.fromIterable(
+ generateTestData(Intervals.DAY, 15 * 60 * 1000, 96)
+ ))
+ on {
+ find("BTC", 2, Date(94 * 15 * 60 * 1000), 15 * 60 * 1000, 2)
+ }.doReturn(Flux.fromIterable(
+ generateTestData(Intervals.DAY * 2, 15 * 60 * 1000, 2)
+ ))
+
+ //24h interval from 00:00pm
+ on {
+ find("BTC", 1, Date(12 * 60 * 60 * 1000), Intervals.DAY, 1)
+ }.doReturn(Flux.fromIterable(
+ generateTestData(Intervals.DAY, Intervals.DAY, 1)
+ ))
+ on {
+ find("BTC", 2, Date(12 * 60 * 60 * 1000), Intervals.DAY, 1)
+ }.doReturn(Flux.fromIterable(
+ generateTestData(Intervals.DAY * 2, Intervals.DAY, 1)
+ ))
+ }
+
+ tickerService = TickerService(repository)
+ }
+
+ //1 min interval
+ @Test
+ fun minuteIntervalTest() {
+
+ val tickers = tickerService
+ .getTickers("BTC", 0L, Intervals.MINUTE, Intervals.DAY / Intervals.MINUTE)
+ .collectList()
+ .block()
+
+ Assertions.assertThat(tickers).hasSize((Intervals.DAY / Intervals.MINUTE).toInt())
+ }
+
+ //24h interval
+ @Test
+ fun dayIntervalTest() {
+
+ val tickers = tickerService
+ .getTickers("BTC", 0L, Intervals.DAY, 3)
+ .collectList()
+ .block()
+
+ Assertions.assertThat(tickers).hasSize(3)
+ }
+
+ //15 min interval
+ @Test
+ fun minute15IntervalTest() {
+
+ val tickers = tickerService
+ .getTickers("BTC", 94 * 15 * 60 * 1000, 15 * 60 * 1000, 100)
+ .collectList()
+ .block()
+
+ Assertions.assertThat(tickers).hasSize(100)
+ }
+
+ //24h interval from 00:00pm
+ @Test
+ fun dayIntervalAfternoonTest() {
+
+ val tickers = tickerService
+ .getTickers("BTC", 12 * 60 * 60 * 1000, Intervals.DAY, 2)
+ .collectList()
+ .block()
+
+ Assertions.assertThat(tickers).hasSize(2)
+ }
+
+ private fun generateTestData(timestampFrom: Long, interval: Long, count: Long): MutableList<CqlTokenTicker> {
+ val tickers = mutableListOf<CqlTokenTicker>()
+
+ for (index in 0 until count) {
+ val timestamp = timestampFrom + interval * index
+
+ tickers.add(
+ CqlTokenTicker(TokenTicker("BTC", timestamp, timestamp + interval, interval))
+ )
+ }
+
+ return tickers
+ }
+
+}