K line daemon & API fix & specs improvement (related to #1689, #1697)

app/api/api_v2/helpers.rb

@@ -27,11 +27,6 @@ def trading_must_be_permitted!
       end
     end
 
-    def redis
-      KlineDB.redis
-    end
-    memoize :redis
-
     def current_user
       # JWT authentication provides member email.
       if env.key?('api_v2.authentic_member_email')
@@ -99,22 +94,5 @@ def format_ticker(ticker)
         }
       }
     end
-
-    def get_k_json
-      key = "peatio:#{params[:market]}:k:#{params[:period]}"
-
-      if params[:timestamp]
-        ts_json = redis.lindex(key, 0)
-        return [] if ts_json.blank?
-        ts = JSON.parse(ts_json).first
-        offset = (params[:timestamp] - ts) / 60 / params[:period]
-        offset = 0 if offset < 0
-        JSON.parse('[%s]' % redis.lrange(key, offset, offset + params[:limit] - 1).join(','))
-      else
-        length = redis.llen(key)
-        offset = [length - params[:limit], 0].max
-        JSON.parse('[%s]' % redis.lrange(key, offset, -1).join(','))
-      end
-    end
   end
 end

app/api/api_v2/k.rb

@@ -8,24 +8,31 @@ class K < Grape::API
     desc 'Get OHLC(k line) of specific market.'
     params do
       use :market
-      optional :limit,     type: Integer, default: 30, values: 1..10000, desc: "Limit the number of returned data points, default to 30."
-      optional :period,    type: Integer, default: 1, values: [1, 5, 15, 30, 60, 120, 240, 360, 720, 1440, 4320, 10080], desc: "Time period of K line, default to 1. You can choose between 1, 5, 15, 30, 60, 120, 240, 360, 720, 1440, 4320, 10080"
-      optional :timestamp, type: Integer, desc: "An integer represents the seconds elapsed since Unix epoch. If set, only k-line data after that time will be returned."
+      optional :period,    type: Integer, default: 1, values: KLineService::AVAILABLE_POINT_PERIODS, desc: "Time period of K line, default to 1. You can choose between #{KLineService::AVAILABLE_POINT_PERIODS.join(', ')}"
+      optional :time_from, type: Integer, desc: "An integer represents the seconds elapsed since Unix epoch. If set, only k-line data after that time will be returned."
+      optional :time_to,   type: Integer, desc: "An integer represents the seconds elapsed since Unix epoch. If set, only k-line data till that time will be returned."
+      optional :limit,     type: Integer, default: 30, values: KLineService::AVAILABLE_POINT_LIMITS, desc: "Limit the number of returned data points default to 30. Ignored if time_from and time_to are given."
     end
     get "/k" do
-      get_k_json
+      KLineService
+        .new(params[:market], params[:period])
+        .get_ohlc(params.slice(:limit, :time_from, :time_to))
     end
 
-    desc "Get K data with pending trades, which are the trades not included in K data yet, because there's delay between trade generated and processed by K data generator."
+    desc "Get K data with pending trades, which are the trades not included in K data yet, because there's delay between trade generated and processed by K data generator.",
+       deprecated: true
     params do
       use :market
       requires :trade_id,  type: Integer, desc: "The trade id of the first trade you received."
-      optional :limit,     type: Integer, default: 30, values: 1..10000, desc: "Limit the number of returned data points, default to 30."
-      optional :period,    type: Integer, default: 1, values: [1, 5, 15, 30, 60, 120, 240, 360, 720, 1440, 4320, 10080], desc: "Time period of K line, default to 1. You can choose between 1, 5, 15, 30, 60, 120, 240, 360, 720, 1440, 4320, 10080"
-      optional :timestamp, type: Integer, desc: "An integer represents the seconds elapsed since Unix epoch. If set, only k-line data after that time will be returned."
+      optional :period,    type: Integer, default: 1, values: KLineService::AVAILABLE_POINT_PERIODS, desc: "Time period of K line, default to 1. You can choose between #{KLineService::AVAILABLE_POINT_PERIODS.join(', ')}"
+      optional :time_from, type: Integer, desc: "An integer represents the seconds elapsed since Unix epoch. If set, only k-line data after that time will be returned."
+      optional :time_to, type: Integer, desc: "An integer represents the seconds elapsed since Unix epoch. If set, only k-line data till that time will be returned."
+      optional :limit,     type: Integer, default: 30, values: KLineService::AVAILABLE_POINT_LIMITS, desc: "Limit the number of returned data points, default to 30."
     end
     get "/k_with_pending_trades" do
-      k = get_k_json
+      k = KLineService
+            .new(params[:market], params[:period])
+            .get_ohlc(params.slice(:limit, :time_from, :time_to))
 
       if params[:trade_id] > 0 && k.present?
         from   = Time.at k.last[0]

app/services/k_line_service.rb

@@ -0,0 +1,97 @@
+# encoding: UTF-8
+# frozen_string_literal: true
+
+class KLineService
+  extend Memoist
+
+  POINT_PERIOD_IN_SECONDS = 60.freeze
+
+  # Point period units are calculated in POINT_PERIOD_IN_SECONDS.
+  # It means that period with value 5 is equal to 5 minutes (5 * POINT_PERIOD_IN_SECONDS = 300).
+  AVAILABLE_POINT_PERIODS = [1, 5, 15, 30, 60, 120, 240, 360, 720, 1440, 4320, 10080].freeze
+
+  AVAILABLE_POINT_LIMITS  = (1..10000).freeze
+
+  def redis
+    Redis.new(
+      url:      ENV.fetch('REDIS_URL'),
+      password: ENV['REDIS_PASSWORD'],
+      db:       1
+    )
+  end
+  memoize :redis
+
+  attr_accessor :market_id, :period
+
+  def initialize(marked_id, period)
+    @market_id = marked_id
+    @period    = period
+  end
+
+  def key
+    "peatio:#{market_id}:k:#{period}"
+  end
+  memoize :key
+
+  # OHCL - open, high, closing, and low prices.
+  def get_ohlc(options={})
+    options = options.symbolize_keys.tap do |o|
+      o.delete(:limit) if o[:time_from].present? && o[:time_to].present?
+    end
+
+    return [] if first_timestamp.nil?
+
+    left_index  = left_index_for(options)
+    right_index = right_index_for(options)
+    return [] if right_index < left_index
+
+    JSON.parse('[%s]' % redis.lrange(key, left_index, right_index).join(','))
+  end
+
+  private
+
+  def points_length
+    redis.llen(key)
+  end
+  memoize :points_length
+
+  def first_timestamp
+    ts_json = redis.lindex(key, 0)
+    ts_json.blank? ? nil : JSON.parse(ts_json).first
+  end
+  memoize :first_timestamp
+
+  def index_for(timestamp)
+    (timestamp - first_timestamp) / POINT_PERIOD_IN_SECONDS / period
+  end
+
+  def left_index_for(options)
+    left_offsets = [0]
+
+    if options[:time_from].present?
+      left_offsets << index_for(options[:time_from])
+    end
+
+    if options[:limit].present?
+      if options[:time_to].present?
+        left_offsets << index_for(options[:time_to]) - options[:limit] + 1
+      elsif options[:time_from].blank?
+        left_offsets << points_length - options[:limit]
+      end
+    end
+    left_offsets.max
+  end
+
+  def right_index_for(options)
+    right_offsets = [points_length]
+
+    if options[:time_to].present?
+      right_offsets << index_for(options[:time_to])
+    end
+
+    if options[:limit].present? && options[:time_from].present?
+      right_offsets << index_for(options[:time_from]) + options[:limit] - 1
+    end
+    right_offsets.min
+  end
+end

lib/daemons/k.rb

@@ -26,8 +26,8 @@ def next_ts(market, period = 1)
   if ts = last_ts(market, period)
     ts += period.minutes
   else
-    if first_trade = Trade.with_market(market).first
-      ts = Trade.with_market(market).first.created_at.to_i
+    if first_trade = Trade.with_market(market).order(created_at: :asc).first
+      ts = first_trade.created_at.to_i
       period == 10080 ? Time.at(ts).beginning_of_week : Time.at(ts -  ts % (period * 60))
     end
   end
@@ -45,7 +45,10 @@ def _k1_set(market, start, period)
 end
 
 def k1(market, start)
-  trades = Trade.with_market(market).where('created_at >= ? AND created_at < ?', start, 1.minutes.since(start)).pluck(:price, :volume)
+  trades = Trade
+             .with_market(market)
+             .where('created_at >= ? AND created_at < ?', start, 1.minutes.since(start))
+             .pluck(:price, :volume)
   return nil if trades.count == 0
 
   prices, volumes = trades.transpose