quik-pipewriter/quik-pipewriter.lua

local alien = require("alien")
json = require "JSON"
date = require "date"
require "string"
struct = require "alien.struct"

w32sleep = alien.kernel32.Sleep
w32sleep:types { ret = "void", abi = "stdcall", "int" }

test_mode = false
running = true
ctx = nil
callback_socket = nil

filter = {
   ["SPBFUT#RIH9"] = true,
   ["SPBFUT#MMH9"] = true,
   ["SPBFUT#USH9"] = true,

   ["SPBFUT#EuH9"] = true,
   ["SPBFUT#EDH9"] = true,
   ["SPBFUT#SiH9"] = true,

   ["SPBFUT#BRJ9"] = true,
   ["SPBFUT#GDH9"] = true,
   
   ["SPBFUT#GZH9"] = true,
   ["SPBFUT#SRH9"] = true,

   ["TQBR#SBER"] = true,
   ["TQBR#GAZP"] = true,
   ["TQBR#LKOH"] = true,
   ["TQBR#NVTK"] = true,
   ["TQBR#MGNT"] = true,
   ["TQBR#GMKN"] = true,
   ["TQBR#ROSN"] = true,
   ["TQBR#VTBR"] = true,
   ["TQBR#SNGS"] = true,
   ["TQBR#MTSS"] = true,
   ["TQBR#ALRS"] = true,
   ["TQBR#TATN"] = true
}

zmq_pollitem = alien.defstruct {
	{"socket", "pointer"},
	{"fd", "int"},
	{"events", "short"},
	{"revents", "short"}
}

zmq_ctx_new = alien.libzmq.zmq_ctx_new
zmq_ctx_new:types { ret = "pointer", abi="stdcall" }

zmq_ctx_term = alien.libzmq.zmq_ctx_term
zmq_ctx_term:types { ret = "int", abi="stdcall", "pointer" }

zmq_connect = alien.libzmq.zmq_connect
zmq_connect:types { ret = "int", abi="stdcall", "pointer", "string" }

zmq_bind = alien.libzmq.zmq_bind
zmq_bind:types { ret = "int", abi="stdcall", "pointer", "string" }

zmq_close = alien.libzmq.zmq_close
zmq_close:types { ret = "int", abi="stdcall", "pointer" }

zmq_disconnect = alien.libzmq.zmq_disconnect
zmq_disconnect:types { ret = "int", abi="stdcall", "pointer", "string" }

zmq_setsockopt = alien.libzmq.zmq_setsockopt
zmq_setsockopt:types { ret = "int", abi="stdcall", "pointer", "int", "pointer", "int" }

zmq_errno = alien.libzmq.zmq_errno
zmq_errno:types { ret = "int", abi="stdcall" }

zmq_msg_close = alien.libzmq.zmq_msg_close
zmq_msg_close:types { ret = "int", abi="stdcall", "pointer" }

zmq_msg_data = alien.libzmq.zmq_msg_data
zmq_msg_data:types { ret = "pointer", abi="stdcall", "pointer" }

zmq_msg_get = alien.libzmq.zmq_msg_get
zmq_msg_get:types { ret = "int", abi="stdcall", "pointer", "int" }

zmq_msg_init_size = alien.libzmq.zmq_msg_init_size
zmq_msg_init_size:types { ret = "int", abi="stdcall", "pointer", "int" }

zmq_msg_init = alien.libzmq.zmq_msg_init
zmq_msg_init:types { ret = "int", abi="stdcall", "pointer" }

zmq_msg_more = alien.libzmq.zmq_msg_more
zmq_msg_more:types { ret = "int", abi="stdcall", "pointer" }

zmq_msg_recv = alien.libzmq.zmq_msg_recv
zmq_msg_recv:types { ret = "int", abi="stdcall", "pointer", "pointer", "int" }

zmq_msg_send = alien.libzmq.zmq_msg_send
zmq_msg_send:types { ret = "int", abi="stdcall", "pointer", "pointer", "int" }

zmq_msg_set = alien.libzmq.zmq_msg_set
zmq_msg_set:types { ret = "int", abi="stdcall", "pointer", "int", "int" }

zmq_msg_size = alien.libzmq.zmq_msg_size
zmq_msg_size:types { ret = "int", abi="stdcall", "pointer" }

zmq_poll = alien.libzmq.zmq_poll
zmq_poll:types { ret = "int", abi="stdcall", "pointer", "int", "int" }

zmq_recv = alien.libzmq.zmq_recv
zmq_recv:types { ret = "int", abi="stdcall", "pointer", "pointer", "int", "int"}

zmq_send = alien.libzmq.zmq_send
zmq_send:types { ret = "int", abi="stdcall", "pointer", "pointer", "int", "int"}

zmq_socket = alien.libzmq.zmq_socket
zmq_socket:types { ret = "pointer", abi="stdcall", "pointer", "int"}

zmq_version = alien.libzmq.zmq_version
zmq_version:types { ret = "void", abi="stdcall", "ref int", "ref int", "ref int"}

zmq_unbind = alien.libzmq.zmq_unbind
zmq_unbind:types { ret = "int", abi="stdcall", "pointer", "string"}

function zmq_msg_new()
	local buf = alien.buffer(64)
	return buf
end

function zmq_msg_as_string(msg)
	return alien.tostring(zmq_msg_data(msg), zmq_msg_size(msg))
end

ZMQ_PAIR = 0
ZMQ_PUB = 1
ZMQ_SUB = 2
ZMQ_REQ = 3
ZMQ_REP = 4
ZMQ_DEALER = 5
ZMQ_ROUTER = 6
ZMQ_PULL = 7
ZMQ_PUSH = 8
ZMQ_XPUB = 9
ZMQ_XSUB = 10
ZMQ_STREAM = 11

ZMQ_POLLIN = 1
ZMQ_POLLOUT = 2
ZMQ_POLLERR = 4

ZMQ_SNDMORE = 2
ZMQ_LINGER = 17

running = true

function to_unixtime(year, month, day)
	return date.diff(date(year, month, day), date(1970, 1, 1)):spanseconds()
end

function to_unixtime_fine(year, month, day, hour, min, sec)
	return date.diff(date(year, month, day, hour, min, sec), date(1970, 1, 1)):spanseconds()
end

function zmq_msg_from_string(str)
	local msg = zmq_msg_new()
	zmq_msg_init_size(msg, str:len())
	buf = zmq_msg_data(msg)
	bsource = alien.buffer(str)
	alien.memcpy(buf, bsource, str:len())
	return msg
end

function zmq_msg_empty()
	local msg = zmq_msg_new()
	zmq_msg_init(msg)
	return msg
end


function send_param_entry(sock, class, sec)
   s = "CT:" ..
	  (class .. "#" .. sec) ..
	  ":" .. 
	  tostring(getParamEx(class, sec, "LAST").param_value) ..
	  ";" ..
	  tostring(getParamEx(class, sec, "VOLTODAY").param_value) ..
	  ";" ..
	  tostring(getParamEx(class, sec, "BID").param_value) ..
	  ";" ..
	  tostring(getParamEx(class, sec, "BIDDEPTH").param_value) ..
	  ";" ..
	  tostring(getParamEx(class, sec, "BIDDEPTHT").param_value) ..
	  ";" ..
	  tostring(getParamEx(class, sec, "OFFER").param_value) ..
	  ";" ..
	  tostring(getParamEx(class, sec, "OFFERDEPTH").param_value) ..
	  ";" ..
	  tostring(getParamEx(class, sec, "OFFERDEPTHT").param_value) ..
	  ";" 

   zmq_msg_send(zmq_msg_from_string(s), callback_socket, 0)
end

function dt_as_string(datetime)
   mcs = 0
   if datetime.mcs ~= nil then
      mcs = datetime.mcs
   end
	return tostring(datetime.year) .. "." ..
		tostring(datetime.month) .. "." ..
		tostring(datetime.day) .. " " ..
		tostring(datetime.hour) .. ":" ..
		tostring(datetime.min) .. ":" ..
		tostring(datetime.sec) .. "." ..
		tostring(mcs)
end

function send_all_trade_entry(sock, ticker_id, price, quantity, datetime, flags)

   s = "AT:" ..
	  (ticker_id) ..
	  ":" ..
	  flags ..
	  ";" ..
	  price ..
	  ";" ..
	  quantity ..
	  ";" ..
	  dt_as_string(datetime) ..
	  ";"
	  

   zmq_msg_send(zmq_msg_from_string(s), callback_socket, 0)
end

function OnParam(class, sec)
   local full_code = class .. "#" .. sec
   if filter[full_code] ~= nil then
	  send_param_entry(sock, class, sec)
   end
end

function OnAllTrade(alltrade)
   local full_code = alltrade.class_code .. "#" .. alltrade.sec_code 
   if filter[full_code] ~= nil then
	  send_all_trade_entry(sock, full_code, alltrade.price, alltrade.qty, alltrade.datetime, alltrade.flags)
   end
end

function OnStop()
   running = false
end

function main()
	local status, err = pcall(pmain)
	if not status then
		message("Error: " .. err)
		PrintDbgStr("Error: " .. err)
	end
	OnStop()
end

function pmain()
   ctx = zmq_ctx_new()
sock = zmq_socket(ctx, ZMQ_PUB)
   local linger = alien.array("int", {0})
   zmq_setsockopt(sock, ZMQ_LINGER, linger.buffer, 4)
   local rc = zmq_bind(sock, "tcp://127.0.0.1:5513")

   if rc ~= 0 then
	  message("Bind error: " .. tostring(rc))
	  return
   end


   sink_socket = zmq_socket(ctx, ZMQ_PULL)
   local rc = zmq_bind(sink_socket, "inproc://events")
   if rc ~= 0 then
	  message("Bind [2] error: " .. tostring(rc))
	  return
   end

   callback_socket = zmq_socket(ctx, ZMQ_PUSH)
   local rc = zmq_connect(callback_socket, "inproc://events")
   if rc ~= 0 then
	  message("Connect [3] error: " .. tostring(rc))
	  return
   end
   

   message("Starting main loop")
   local last_action = os.time()
   while running do
	  local pi = zmq_pollitem:new()
	  pi.socket = sink_socket
	  pi.fd = 0
	  pi.events = ZMQ_POLLIN
	  pi.revents = 0
	  rc = zmq_poll(pi(), 1, 5000) 
	  if rc > 0 and pi.revents == ZMQ_POLLIN then
		 local msg = zmq_msg_new()
		 zmq_msg_init(msg)
		 rc = zmq_msg_recv(msg, sink_socket, 0) 
		 if rc == -1 then
			running = false
			message("rc: " .. tostring(rc) .. "; " .. tostring(zmq_errno()))
		 else
			zmq_msg_send(msg, sock, 0)
		 end
	  end

	  if test_mode then
		if os.time() - last_action > 10 then
			message("Sending 500 ticks")
			for i=1,500 do
			 alltrade = {
				["class_code"] = "SPBFUT",
				["sec_code"] = "RIH9",
				["price"] = i,
				["qty"] = 1,
				["datetime"] = os.date("*t"),
				["flags"] = 1
			 }
			 
				last_action = os.time()
				OnAllTrade(alltrade)
			 end
		 end
	  end
   end
   zmq_close(sock)
   zmq_close(sink_socket)
   zmq_close(callback_socket)
   zmq_ctx_term(ctx)
end