Class: Irix::Bitfinex

Inherits:
Peatio::Upstream::Base
  • Object
show all
Defined in:
lib/irix/bitfinex.rb

Constant Summary collapse

MIN_INCREMENT_COUNT_TO_SNAPSHOT = 
100
MIN_PERIOD_TO_SNAPSHOT = 
5
MAX_PERIOD_TO_SNAPSHOT = 
60

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config) ⇒ Bitfinex

Returns a new instance of Bitfinex.



14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/irix/bitfinex.rb', line 14

def initialize(config)
  super
  @connection = Faraday.new(url: (config['rest']).to_s) do |builder|
    builder.response :json
    builder.response :logger if config['debug']
    builder.adapter(@adapter)
    unless config['verify_ssl'].nil?
      builder.ssl[:verify] = config['verify_ssl']
    end
  end
  @open_channels = {}
  @ping_set = false
  @rest = (config['rest']).to_s
  @ws_url = (config['websocket']).to_s
end

Instance Attribute Details

#asks ⇒ Object

Returns the value of attribute asks.



11
12
13
 
# File 'lib/irix/bitfinex.rb', line 11

def asks
  @asks
end

#bids ⇒ Object

Returns the value of attribute bids.



11
12
13
 
# File 'lib/irix/bitfinex.rb', line 11

def bids
  @bids
end

#increment_count ⇒ Object

Returns the value of attribute increment_count.



11
12
13
 
# File 'lib/irix/bitfinex.rb', line 11

def increment_count
  @increment_count
end

#open_channels ⇒ Object

Returns the value of attribute open_channels.



11
12
13
 
# File 'lib/irix/bitfinex.rb', line 11

def open_channels
  @open_channels
end

#sequence_number ⇒ Object

Returns the value of attribute sequence_number.



11
12
13
 
# File 'lib/irix/bitfinex.rb', line 11

def sequence_number
  @sequence_number
end

#snap ⇒ Object

Returns the value of attribute snap.



11
12
13
 
# File 'lib/irix/bitfinex.rb', line 11

def snap
  @snap
end

#snapshot_time ⇒ Object

Returns the value of attribute snapshot_time.



11
12
13
 
# File 'lib/irix/bitfinex.rb', line 11

def snapshot_time
  @snapshot_time
end

Instance Method Details

#detect_order(msg) ⇒ Object

[

CHANNEL_ID,
[
  PRICE,
  COUNT,
  AMOUNT
]

]



125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/irix/bitfinex.rb', line 125

def detect_order(msg)
  if msg[1][0].is_a?(Array)
    msg[1].each do |point|
      if point[2] > 0
        @snap['bids'] << [point[0].to_s, point[2].to_s]
      else
        @snap['asks'] << [point[0].to_s, point[2].abs.to_s]
      end
    end
    publish_snapshot
  else
    if @increment_count < MIN_INCREMENT_COUNT_TO_SNAPSHOT && @snapshot_time <= Time.now - MAX_PERIOD_TO_SNAPSHOT
      publish_snapshot
      @increment_count = 0
    elsif @increment_count >= MIN_INCREMENT_COUNT_TO_SNAPSHOT && @snapshot_time < Time.now - MIN_PERIOD_TO_SNAPSHOT
      publish_snapshot
      @increment_count = 0
    end

    fill_increment(msg[1])
  end
end

#detect_trade(msg) ⇒ Object 



102
103
104
105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/irix/bitfinex.rb', line 102

def detect_trade(msg)
  if msg[1] == 'tu'
    data = msg[2]
    trade =
      {
        'tid' => data[0],
        'amount' => data[2].to_d.abs,
        'price' => data[3],
        'date' => data[1] / 1000,
        'taker_type' => data[2].to_d.positive? ? 'buy' : 'sell'
      }
    notify_public_trade(trade)
  end
end

#fill_increment(order) ⇒ Object 



148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
 
# File 'lib/irix/bitfinex.rb', line 148

def fill_increment(order)
  side = order[2].positive? ? 'bid' : 'ask'
  price = order[0].to_s
  if order[1].zero?
    amount = 0
    @snap["#{side}s"].delete_if { |point| point[0] == price }
  else
    amount = order[2].abs.to_s
    @snap["#{side}s"].delete_if { |point| point[0] == price }
    @snap["#{side}s"] << [price.to_s, amount.to_s]
  end
  if side == 'bid'
    @bids.delete_if { |point| point[0] == price }
    @bids << [price.to_s, amount.to_s]
  elsif side == 'ask'
    @asks.delete_if { |point| point[0] == price }
    @asks << [price.to_s, amount.to_s]
  end
  @increment_count += 1
end

#info(msg) ⇒ Object 



201
202
203
 
# File 'lib/irix/bitfinex.rb', line 201

def info(msg)
  Rails.logger.info "Bitfinex: #{msg}"
end

#message_event(msg) ⇒ Object 



191
192
193
194
195
196
197
198
199
 
# File 'lib/irix/bitfinex.rb', line 191

def message_event(msg)
  case msg['event']
  when 'subscribed'
    Rails.logger.info "Event: #{msg}"
    @open_channels[msg['chanId']] = msg['channel']
  when 'error'
    Rails.logger.info "Event: #{msg} ignored"
  end
end

#publish_increment ⇒ Object 



169
170
171
172
173
174
175
176
177
178
179
180
181
 
# File 'lib/irix/bitfinex.rb', line 169

def publish_increment
  inc = {}
  inc['bids'] = @bids.sort.reverse if @bids.present?
  inc['asks'] = @asks.sort if @asks.present?
  if inc.present?
    @sequence_number += 1
    @peatio_mq.enqueue_event('public', @market, 'ob-inc',
                             'bids' => inc['bids'], 'asks' => inc['asks'],
                             'sequence' => @sequence_number)
  end
  @bids = []
  @asks = []
end

#publish_snapshot ⇒ Object 



183
184
185
186
187
188
189
 
# File 'lib/irix/bitfinex.rb', line 183

def publish_snapshot
  @snapshot_time = Time.now
  @peatio_mq.enqueue_event('public', @market, 'ob-snap',
                           'bids' => @snap['bids'].sort.reverse,
                           'asks' => @snap['asks'].sort,
                           'sequence' => @sequence_number)
end

#subscribe_orderbook(market, ws) ⇒ Object 



63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/irix/bitfinex.rb', line 63

def subscribe_orderbook(market, ws)
  return unless @config['orderbook_proxy']

  @sequence_number = 0
  @increment_count = 0
  @bids = []
  @asks = []
  @snap = { 'asks' => [], 'bids' => [] }
  sub = {
    event: 'subscribe',
    channel: 'book',
    symbol: market.upcase,
    len: 25
  }
  Rails.logger.info 'Open event' + sub.to_s
  EM.next_tick do
    ws.send(JSON.generate(sub))
  end
  Fiber.new do
    EM::Synchrony.add_periodic_timer(0.2) do
      publish_increment
    end
  end.resume
end

#subscribe_trades(market, ws) ⇒ Object 



48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/irix/bitfinex.rb', line 48

def subscribe_trades(market, ws)
  return unless @config['trade_proxy']

  sub = {
    event: 'subscribe',
    channel: 'trades',
    symbol: market.upcase
  }

  Rails.logger.info 'Open event' + sub.to_s
  EM.next_tick do
    ws.send(JSON.generate(sub))
  end
end

#ws_connect ⇒ Object 



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/irix/bitfinex.rb', line 30

def ws_connect
  super
  return if @ping_set

  @ws.on(:open) do |_e|
    subscribe_trades(@target, @ws)
    subscribe_orderbook(@target, @ws)
    logger.info { 'Websocket connected' }
  end

  Fiber.new do
    EM::Synchrony.add_periodic_timer(80) do
      @ws.send('{"event":"ping"}')
    end
  end.resume
  @ping_set = true
end

#ws_read_public_message(msg) ⇒ Object 



88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/irix/bitfinex.rb', line 88

def ws_read_public_message(msg)
  if msg.is_a?(Array)
    if msg[1] == 'hb'
      @ws.send('{"event":"ping"}')
    elsif @open_channels[msg[0]] == 'trades'
      detect_trade(msg)
    elsif @open_channels[msg[0]] == 'book'
      detect_order(msg)
    end
  elsif msg.is_a?(Hash)
    message_event(msg)
  end
end