Tasks Scheduler (Refactor)

tasks_spike.rb

require 'rufus/scheduler'
require 'json'
require 'httparty'
require 'rake'

##################################################################
### INITIALIZERS
##################################################################
@start_day = Date.today.beginning_of_day
@finish_day = Date.today.end_of_day

@indicators = Rbandit::Trdindopt.all
@exchanges = Rbandit::Exchopt.all

@last_transaction = 0

@first_trdopt_row = Rbandit::Trdopt.where(:ts => @start_day..@finish_day).first
@first_trdstatsopt_row = Rbandit::Trdstatsopt.where(:ds => Date.today).first

@last_trdopt_row = []
@last_trdopt_row_underlying = []
@last_trdstatsopt_row = []

main = Rufus::Scheduler.start_new
secondary = Rufus::Scheduler.start_new
@underlying_hash = Hash.new(0)
@underlying_last_row = 0
@underlying_started = 0
@underlying_array_complete = []

##################################################################
### METHODS
##################################################################
def get_ind_name(ind)
  if ind == "TOTAL"
    return ind
  else
    @indicators.find { |i| i[:id] == ind }.name
  end
end

def get_exch_name(exch)
  @exchanges.find { |e| e[:id] == exch }.name
end

def get_callput(letter)
  if letter == "P"
    "PUT"
  elsif letter == "C"
    "CALL"
  else
    "UNKN"
  end
end

##################################################################
### SCHEDULER
##################################################################
if defined?(Rails::Console)
  puts "You're in the console. Not running scheduler..."
else
  main.every '5s', allow_overlapping: false do
    @last_trdopt_row = Rbandit::Trdopt.last
    @last_trdstatsopt_row = Rbandit::Trdstatsopt.last

    #################################################
    ##### INDICATORS
    #################################################
      ind_count = Rbandit::Trdopt.where(:id => @first_trdopt_row.id..@last_trdopt_row.id).count(group: 'ind')
      sorted_ind_count = (ind_count.sort { |a,b| a[1] <=> b[1] }).reverse
      ind_count = Hash[sorted_ind_count]
      ind_count["TOTAL"] = ind_count.values.sum

      ind_array = []

      ind_count.each do |k,v|
        input_hash = Hash.new
        input_hash["label"] = get_ind_name(k)
        input_hash["value"] = v
        ind_array << input_hash 
      end

      puts "Sending Indicators!"
      HTTParty.post('http://localhost:3030/widgets/indicators_list', :body => { auth_token: "YOUR_AUTH_TOKEN", items: ind_array }.to_json)

    #################################################
    ##### TRANSACTIONS EXECUTED
    #################################################
      HTTParty.post('http://localhost:3030/widgets/total_transactions', :body => { auth_token: "YOUR_AUTH_TOKEN", current: ind_count["TOTAL"], last: @last_transaction }.to_json)

      @last_transaction = ind_count["TOTAL"]
    #################################################
    ##### EXCHANGES
    #################################################
      exch_count = Rbandit::Trdopt.where(:id => @first_trdopt_row.id..@last_trdopt_row.id).count(group: 'exchcode')
      sorted_exch_count = (exch_count.sort { |a,b| a[1] <=> b[1] })
      exch_count = Hash[sorted_exch_count]
      exch_names = exch_count.keys
      exch_names.map! { |ex| get_exch_name(ex) }

      HTTParty.post('http://localhost:3030/widgets/highbar_ind', :body => { auth_token: "YOUR_AUTH_TOKEN", series: [{ data: exch_count.values }], categories: exch_names, color: '#efad1b' }.to_json)

    #################################################
    ##### TOP 10 INSTRUMENTS BY VOLUME
    #################################################
      top_insts = Rbandit::Trdstatsopt.includes(:instropt).where(:id => @first_trdstatsopt_row.id..@last_trdstatsopt_row.id).order('volume DESC').limit(10)

      top_insts_cats = []
      top_insts_data = []
      
      top_insts.each do |list|
        top_insts_cats << "#{list.instropt.underlying} - #{list.instropt.expiration} - #{list.instropt.strike.to_f} - #{get_callput(list.instropt.callput)}"
        top_insts_data << list.volume
      end

      HTTParty.post('http://localhost:3030/widgets/instruments', :body => { auth_token: "YOUR_AUTH_TOKEN", series: [{ name: 'Instruments', data: top_insts_data }], categories: top_insts_cats, color: '#2c3e50' }.to_json)
    
    #################################################
    ##### RECENT TRADES
    #################################################
    recent_trades = Rbandit::Trdopt.includes(:instropt).where(:id => @last_trdopt_row.id-3..@last_trdopt_row.id)

    recent_trades_array = []

    recent_trades.each do |trade|
      input_hash = Hash.new
      input_hash["label"] = "[#{ trade.ts.strftime("%I:%M:%S %p") }] - <b>#{ trade.instropt.underlying }</b> | #{ trade.instropt.expiration } | Strike: $#{ trade.instropt.strike.to_f } | #{ get_callput(trade.instropt.callput) }"
      input_hash["value"] = "@ Price: $#{ trade.price.to_f} | Size: #{ trade.size } | Exchange: #{ get_exch_name(trade.exchcode) } | Indicator: #{get_ind_name(trade.ind)}"
      recent_trades_array << input_hash
    end

    recent_trades_array.reverse!

    HTTParty.post('http://localhost:3030/widgets/recent_trades_list', :body => { auth_token: "YOUR_AUTH_TOKEN", items: recent_trades_array }.to_json)

    #################################################
    ##### TOP 5 UNDERLYING
    #################################################

    #top_underlying = Rbandit::Trdopt.includes(:instropt).where(:id => @first_trdopt_row.id..@last_trdopt_row.id).count

    #################################################
    ##### TOP 5 SIZES
    #################################################

    top_sizes = Rbandit::Trdopt.includes(:instropt).where(:id => @first_trdopt_row.id..@last_trdopt_row.id).order("size DESC").limit(10)

    size_sample = top_sizes.map do |s|
      ["#{ s.instropt.underlying } - #{ s.instropt.expiration } @ $#{ s.instropt.strike.to_f } - #{ get_callput(s.instropt.callput) }", s.size.to_i]
    end

    size_sample.uniq!(&:first)

    sizes_data = size_sample.map { |data| data[1] }
    sizes_cats = size_sample.map { |cat| cat[0] }

    # sizes_cats = top_sizes.map do |s| 
    #   "#{ s.instropt.underlying } - #{ s.instropt.expiration } @ $#{ s.instropt.strike.to_f } - #{ get_callput(s.instropt.callput) }"
    # end

    # sizes_data = top_sizes.map { |s| s.size.to_i }

    HTTParty.post('http://localhost:3030/widgets/top5_sizes', :body => { auth_token: "YOUR_AUTH_TOKEN", series: [{ name: 'Instruments', data: sizes_data }], categories: sizes_cats, color: '#95a5a6' }.to_json)

    puts "#{@underlying_hash.values.sum}"
  end

  secondary.every '15s', allow_overlapping: false do

    if @underlying_started == 0
      puts "Starting underlying calculation..."

      @last_trdopt_row_underlying = @last_trdstatsopt_row.id

      Rbandit::Trdopt.includes(:instropt).where(:id => @first_trdopt_row.id..@last_trdopt_row_underlying).find_each(:batch_size => 5000) do |t|
          #puts "#{t.instropt.underlying}: #{@underlying_hash.values.sum}"
          @underlying_hash[t.instropt.underlying] += 1
      end

      underlying_hash2 = @underlying_hash.sort_by &:last
      underlying_hash2.reverse!

      underlying_array = []

      underlying_hash2[0..4].each do |list|
        input_hash = Hash.new
        input_hash["name"] = list[0]
        input_hash["data"] = [list[1]]
        input_hash["pointWidth"] = 500
        underlying_array << input_hash
      end

      HTTParty.post('http://localhost:3030/widgets/stacker_symbols', :body => { auth_token: "YOUR_AUTH_TOKEN", series: underlying_array, color: '#d35400'}.to_json)

      puts "Underlying INITALIZED!"

      @underlying_array_complete = underlying_array
      @underlying_started += 1
      @underlying_last_row = @last_trdopt_row_underlying
      @last_trdopt_row_underlying = @last_trdopt_row.id
    elsif @underlying_last_row != @last_trdopt_row_underlying

      puts "UNDERLYING LAST: #{@underlying_last_row} - #{@last_trdopt_row_underlying}"
      Rbandit::Trdopt.includes(:instropt).where(:id => @underlying_last_row+1..@last_trdopt_row_underlying).find_each(:batch_size => 5000) do |t|
          @underlying_hash[t.instropt.underlying] += 1
      end

      underlying_hash2 = @underlying_hash.sort_by &:last
      underlying_hash2.reverse!

      underlying_array = []

      underlying_hash2[0..4].each do |list|
        input_hash = Hash.new
        input_hash["name"] = list[0]
        input_hash["data"] = [list[1]]
        input_hash["pointWidth"] = 500
        underlying_array << input_hash
      end

       @under_array_complete = underlying_array

      HTTParty.post('http://localhost:3030/widgets/stacker_symbols', :body => { auth_token: "YOUR_AUTH_TOKEN", series: underlying_array, color: '#d35400'}.to_json)
      puts "UNDERLYING UPDATED!"
      @underlying_last_row = @last_trdopt_row_underlying
      @last_trdopt_row_underlying = @last_trdopt_row.id
    else
      "Nothing to see here."

      HTTParty.post('http://localhost:3030/widgets/stacker_symbols', :body => { auth_token: "YOUR_AUTH_TOKEN", series: @underlying_array_complete, color: '#d35400'}.to_json)
    end
  end
end