class BitstampStreamingMarketDataService implements StreamingMarketDataService {
  private final BitstampStreamingService service;

  public BitstampStreamingMarketDataService(BitstampStreamingService service) {
    this.service = service;
  }

  public Observable<OrderBook> getFullOrderBook(CurrencyPair currencyPair, Object... args) {
    return getOrderBook("diff_order_book", currencyPair, args);
  }

  @Override
  public Observable<OrderBook> getOrderBook(CurrencyPair currencyPair, Object... args) {
    return getOrderBook("order_book", currencyPair, args);
  }

  private Observable<OrderBook> getOrderBook(
      String channelPrefix, CurrencyPair currencyPair, Object... args) {
    String channelName = channelPrefix + getChannelPostfix(currencyPair);

    return service
        .subscribeChannel(channelName, BitstampStreamingService.EVENT_ORDERBOOK)
        .map(
            s -> {
              ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();
              BitstampOrderBook orderBook =
                  mapper.treeToValue(s.get("data"), BitstampOrderBook.class);
              return BitstampAdapters.adaptOrderBook(orderBook, currencyPair);
            });
  }

  @Override
  public Observable<Ticker> getTicker(CurrencyPair currencyPair, Object... args) {
    return getOrderBook(currencyPair, args)
        .map(orderBook -> mapOrderBookToTicker(currencyPair, orderBook));
  }

  private Ticker mapOrderBookToTicker(CurrencyPair currencyPair, OrderBook orderBook) {<FILL_FUNCTION_BODY>}

  @Override
  public Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args) {
    String channelName = "live_trades" + getChannelPostfix(currencyPair);

    return service
        .subscribeChannel(channelName, BitstampStreamingService.EVENT_TRADE)
        .map(
            s -> {
              ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();
              BitstampWebSocketTransaction transactions =
                  mapper.treeToValue(s.get("data"), BitstampWebSocketTransaction.class);
              return BitstampAdapters.adaptTrade(transactions, currencyPair, 1);
            });
  }

  private String getChannelPostfix(CurrencyPair currencyPair) {
    return "_"
        + currencyPair.base.toString().toLowerCase()
        + currencyPair.counter.toString().toLowerCase();
  }
}

    final LimitOrder ask = orderBook.getAsks().get(0);
    final LimitOrder bid = orderBook.getBids().get(0);

    return new Ticker.Builder()
        .instrument(currencyPair)
        .bid(bid.getLimitPrice())
        .bidSize(bid.getOriginalAmount())
        .ask(ask.getLimitPrice())
        .askSize(ask.getOriginalAmount())
        .timestamp(orderBook.getTimeStamp())
        .build();
  
<no_super_class>