Template Perdagangan Frekuensi Tinggi Websocket OKX & Binance (Beragam Jenis)
Tanggal Pembuatan:
2022-12-27 15:37:42
Akhirnya memodifikasi:
2024-01-02 16:36:13
menyalin:
75
Jumlah klik:
4507
QCoder
11
fokus pada
607
Pengikut
Fungsi implementasi:
- Subscription, trades dan deep synthesis
- Pendaftaran aliran pribadi, pemeliharaan surat perintah dan posisi lokal, dan informasi saldo
- Mempertahankan status amendOrders untuk mengubah pesanan
- Mendukung berbagai varietas, memudahkan strategi frekuensi tinggi untuk mengoperasikan beberapa pasar secara bersamaan dengan latensi minimum
- Tambahkan fungsi EventLoop (diperlukan dukungan host terbaru), fungsi ini dapat digunakan dalam kasus koneksi beberapa websocket, menunggu data websocket arbitrer untuk dikembalikan, menghindari polling
- Dukungan OKX dan Binance Futures dan Binance Cash and Cash Leveraged Trading, OKX Cash belum didukung, yang berminat dapat mengubahnya sendiri
- OKX Streaming Private Subscription membutuhkan KEY dan password yang telah dibuat sebagai parameter yang ditulis dalam template
- Mengatur fungsi jaringan dasar seperti menghubungkan kembali kabel yang putus
- Pengembalian acara yang mendukung pemesanan order streaming pribadi untuk membatalkan pengembalian
- Untuk menyamakan logika pesanan dari dua bursa, lihat template sumber atau contoh
- Websocket interface yang mendukung berlangganan ke beberapa bursa sekaligus
- Modifikasi dan penghapusan pesanan dapat dilakukan secara massal sebagai pengembalian tugas, yang akan secara otomatis memproses tugas transaksi
Pada saat ordering frekuensi tinggi, order yang dikembalikan oleh bursa tidak akan tepat waktu (saluran pribadi), tetapi pada saat ini situasi mungkin telah berubah, kita perlu mempertahankan status order secara lokal, strategi telah terwujud, dan kita tidak akan dapat melakukan hal-hal seperti itu. Template akan secara otomatis mengidentifikasi apakah itu OKX atau Binance, berikut adalah contoh sederhana, sangat disarankan untuk membaca kode sumber sehingga Anda dapat menyesuaikan fungsionalitas Anda sendiri dengan lebih dalam
Berikut ini adalah contoh sederhana.
function onTick(ctx, event) {
if (event.depth) {
Log("depth update", event)
}
if (event.trades) {
Log("trades received", event)
}
if (event.balance || event.positions) {
Log("account update", event)
}
if (event.orders) {
Log("private orders update", event)
}
// include orders, positions, balance for latest
Log("account", ctx.wsPrivate.account)
// for test only
if (false) {
return {
amendOrders: [{
instId: event.instId,
clOrdId: "xxxx*****",
cxlOnFail: true,
newSz: "2",
}],
newOrders: [{
instId: event.instId,
clOrdId: UUID(),
side: "sell",
tdMode: "cross",
ordType: "post_only",
px: event.depth.asks[0].price.toFixed(4),
sz: "1",
}, {
instId: event.instId,
clOrdId: UUID(),
side: "sell",
tdMode: "cross",
ordType: "post_only",
px: event.depth.asks[0].price.toFixed(4),
sz: "1",
}],
cancelOrders: [{
instId: order.instId,
clOrdId: order.Id
}]
}
}
}
function main() {
let instId = exchange.SetContractType("swap").InstrumentID
let ctx = $.NewWSS(exchange, function(ws) {
let msg = null
if (exchange.GetName() === 'Futures_OKCoin') {
msg = {
op: "subscribe",
args: [{
channel: "books5",
instId: instId,
}, {
channel: "trades",
instId: instId,
}]
}
} else {
let symbol = exchange.GetCurrency().replace('_', '').toLowerCase()
msg = {
method: "SUBSCRIBE",
params: [symbol + "@aggTrade", symbol + "@depth20@100ms"],
id: "1",
}
}
ws.write(JSON.stringify(msg))
Log("subscribe", msg, "channel")
LogStatus("Ready")
}, onTick, Debug)
while (true) {
ctx.poll()
EventLoop(1000)
}
}
Ini berjalan seperti ini, Anda juga dapat berlangganan beberapa varietas dalam fungsi berlangganan
Berikut ini adalah hasil dari strategi market maker umum OKX dan Bitcoin berdasarkan template ini.
Kode Sumber Strategi
/* jshint esversion: 6 */
// AccessKey and Phassphrase only need if OKX
$.NewWSS = function(e, onWSSLogin, onWSSTick, Debug, UseMargin, AccessKey, Passphrase) {
function NewOKXWebSocketPrivate(ctx) {
let self = {
ws: null,
isReadyDic: {},
lastPing: 0,
ctx: ctx,
account: {
orders: {},
cancelPending: {},
ordersCount: {
pending: 0,
buy: 0,
amend: 0,
canceled: 0,
filled: 0,
push: 0,
sell: 0
},
positions: {},
balance: {},
init_balance: {},
balanceUpdate: 0,
positionsUpdate: 0,
ordersUpdate: 0
}
}
let acc = _C(ctx.e.GetAccount)
let pair = ctx.e.GetCurrency().split('_')
self.account.balance[pair[1]] = {
free: acc.Balance,
borrowed: 0,
locked: 0
}
self.account.balance[pair[0]] = {
free: acc.Stock,
borrowed: 0,
locked: 0
}
self.account.init_balance[pair[1]] = {
free: acc.Balance,
borrowed: 0,
locked: 0
}
self.account.init_balance[pair[0]] = {
free: acc.Stock,
borrowed: 0,
locked: 0
}
self.reset = function() {
if (self.ws) {
self.ws.close()
self.ws = null
}
self.isReadyDic = {}
}
self.send = function(msg) {
if (!self.ws) {
return
}
self.ws.write(JSON.stringify(msg))
}
self.getOrdersMap = function(instId) {
let mp = self.account.orders[instId]
if (typeof(mp) === 'undefined') {
mp = {}
self.account.orders[instId] = mp
}
return mp
}
self.batchOrders = function(orders) {
for (let i = 0; i < orders.length; i += 20) {
self.send({
"id": "batchOrders",
"op": "batch-orders",
"args": orders.slice(i, i + 20)
})
if (Debug) {
Log("batchOrders", orders.slice(i, i + 20))
}
}
let ts = new Date().getTime()
orders.forEach(function(item) {
let mp = self.getOrdersMap(item.instId)
mp[item.clOrdId] = {
Id: item.clOrdId,
Created: ts,
instId: item.instId,
Price: Number(item.px),
Amount: Number(item.sz),
Type: item.side == "sell" ? ORDER_TYPE_SELL : ORDER_TYPE_BUY,
DealAmount: 0,
Status: ORDER_STATE_PENDING
}
})
}
self.cancelOrders = function(orders) {
let ts = new Date().getTime()
orders.forEach(function(item) {
if (typeof(self.account.cancelPending[item.clOrdId]) === 'undefined') {
self.account.cancelPending[item.clOrdId] = {
retry: 0,
ts: ts,
data: item
}
}
})
// remove from orders
orders.forEach(function(item) {
delete self.getOrdersMap(item.instId)[item.clOrdId]
})
for (let id in self.account.cancelPending) {
let item = self.account.cancelPending[id]
if (ts - item.ts > 10000) {
// recancel
orders.push(item.data)
Log("remove timeout order", item, "#ff0000")
item.retry += 1
item.ts = ts
}
if (item.retry > 10) {
Log("force remove order", item, "#ff0000")
delete self.account.cancelPending[id]
}
}
for (let i = 0; i < orders.length; i += 20) {
self.send({
"id": "cancelOrders",
"op": "batch-cancel-orders",
"args": orders.slice(i, i + 20)
})
}
if (Debug) {
Log("cancelOrders", orders)
}
}
self.amendOrders = function(orders) {
let ts = new Date().getTime()
orders.forEach(function(item) {
let order = self.getOrdersMap(item.instId)[item.clOrdId]
if (order) {
order.Price = Number(item.newPx)
order.Amount = Number(item.newSz)
order.Update = undefined
order.Created = ts
}
})
self.send({
"id": "amendOrders",
"op": "batch-amend-orders",
"args": orders,
})
if (Debug) {
Log("amendOrders", orders)
}
}
self.processOrders = function(data, reset) {
let ret = []
if (Debug) {
Log("ORDERS: ", data)
}
if (reset) {
self.account.orders = {}
for (let k in self.account.ordersCount) {
self.account.ordersCount[k] = 0
}
}
data.forEach(function(item) {
let mp = self.getOrdersMap(item.instId)
let dataOrder = {
Id: item.clOrdId,
instId: item.instId,
Info: item,
Price: Number(item.px),
Amount: Number(item.sz),
Update: Number(item.uTime || item.cTime),
Created: Number(item.cTime),
Type: item.side == "buy" ? ORDER_TYPE_BUY : ORDER_TYPE_SELL,
DealAmount: Number(item.accFillSz)
}
if (item.state == "canceled") {
dataOrder.Status = ORDER_STATE_CANCELED
self.account.ordersCount.canceled += 1
self.account.ordersCount.pending -= 1
} else if (item.state == "live") {
dataOrder.Status = ORDER_STATE_PENDING
if (item.amendResult == "0") {
self.account.ordersCount.amend += 1
} else {
self.account.ordersCount.push += 1
self.account.ordersCount.pending += 1
}
} else if (item.state == "partially_filled") {
dataOrder.Status = ORDER_STATE_PENDING
} else if (item.state == "filled") {
dataOrder.Status = ORDER_STATE_CLOSED
self.account.ordersCount.filled += 1
self.account.ordersCount.pending -= 1
} else {
throw "unknow order state " + JSON.stringify(item)
}
// remove from cancelPending orders
if (dataOrder.Status != ORDER_STATE_PENDING) {
delete self.account.cancelPending[dataOrder.Id]
if (Debug) {
Log("remove from cancel pending orders", dataOrder)
}
}
// update anyway
let oldOrder = mp[dataOrder.Id]
if ((!oldOrder) || (dataOrder.DealAmount != oldOrder.DealAmount)) {
if (item.state == "partially_filled" || item.state == "filled") {
Log(item.state, item.side, dataOrder, (item.clOrdId.indexOf('YYYYY') == 0 ? '#ff0000' : ''))
}
}
let update = false
if (dataOrder.Status == ORDER_STATE_PENDING) {
// 修改订单的命令还没返回就收到了new订单的update导致时间序列错乱
if (oldOrder) {
update = true
Object.assign(oldOrder, dataOrder)
} else {
mp[dataOrder.Id] = dataOrder
update = true
}
if (Debug) {
let suffix = ''
if (self.ctx.depth) {
suffix = 'bid:' + JSON.stringify(self.ctx.depth.bids[0]) + ', ask:' + JSON.stringify(self.ctx.depth.asks[0])
}
Log("update order", Boolean(oldOrder), mp[dataOrder.Id], item, suffix)
}
} else {
update = true
if (oldOrder) {
// avoid ref
oldOrder.Status = dataOrder.Status
delete mp[dataOrder.Id]
}
if (Debug) {
Log("order " + item.state, dataOrder)
}
}
if (update) {
ret.push(dataOrder)
}
self.account.ordersUpdate = Number(item.uTime || item.cTime)
})
return ret
}
self.processMsg = function(msg) {
if (Debug) {
Log("MSG:", msg)
}
let obj = JSON.parse(msg)
if (obj.event == "error") {
Log("Error:", obj.msg)
Sleep(1000)
self.ws.close()
} else if (obj.event == "login") {
Log("Login success")
self.ws.write(JSON.stringify({
"op": "subscribe",
"args": [{
"channel": "balance_and_position"
}, {
"channel": "orders",
"instType": "ANY"
}]
}))
} else if (obj.event == "subscribe") {
Log("subscribe OK", obj.arg.channel)
if (obj.arg.channel == "orders" && !self.isReadyDic["orders"]) {
let ret = self.ctx.e.IO("api", "GET", "/api/v5/trade/orders-pending")
if (ret && ret.code == "0") {
self.processOrders(ret.data, true)
Log("pocess orders ok", self.account.ordersCount)
} else {
Log("process order failed", ret, "#ff0000")
}
self.isReadyDic["orders"] = true
}
} else if (obj && obj.arg && obj.data && obj.data.length > 0) {
let event = {}
if (obj.arg.channel == 'balance_and_position') {
if (Debug) {
Log(obj, "#ff0000")
}
event.ts = Number(obj.data[0].pTime)
if (obj.data[0].posData) {
let positions = {}
obj.data[0].posData.forEach(function(item) {
if (typeof(positions[item.instId]) === 'undefined') {
positions[item.instId] = 0
}
positions[item.instId] = {
Amount: Number(item.pos),
Price: Number(item.avgPx)
}
self.account.positionsUpdate = Number(item.uTime)
})
for (let instId in positions) {
self.account.positions[instId] = positions[instId]
}
event.positions = positions
}
if (obj.data[0].balData) {
let balance = {}
obj.data[0].balData.forEach(function(item) {
balance[item.ccy] = {
free: Number(item.cashBal),
locked: 0
}
self.account.balanceUpdate = Number(item.uTime)
})
for (let instId in balance) {
self.account.balance[instId] = balance[instId]
}
event.balance = balance
}
self.isReadyDic["account"] = true
} else if (obj.arg.channel == 'orders') {
event.orders = self.processOrders(obj.data)
event.ts = Number(obj.data[0].uTime)
} else {
// {"id":"amendOrders","op":"batch-amend-orders" ...}
//Log("DATA RECV", "<" + msg + ">")
}
// position change
if (event.ts) {
self.ctx.processTick(event)
}
} else {
//Log("RECV", "<" + msg + ">")
}
}
self.poll = function(timeout) {
if (typeof(AccessKey) === 'undefined' || AccessKey.length == 0) {
return
}
let ts = new Date().getTime()
if (self.lastPing == 0) {
self.lastPing = ts
}
if (self.ws == null) {
self.ws = Dial("wss://ws.okx.com:8443/ws/v5/private")
if (self.ws) {
let tsStr = (ts / 1000).toString()
let authMsg = {
"op": "login",
"args": [{
"apiKey": AccessKey,
"passphrase": Passphrase,
"timestamp": tsStr,
"sign": self.ctx.e.HMAC("sha256", "base64", tsStr + "GET" + "/users/self/verify", "{{secretkey}}")
}]
}
self.ws.write(JSON.stringify(authMsg))
}
}
if (!self.ws) {
return;
}
if (ts - self.lastPing > 10000) {
self.ws.write("ping")
self.lastPing = ts
}
let lastRead = false
while (true) {
let msg = self.ws.read(-1)
if (msg == "") {
self.reset()
break
}
if (msg == null) {
if (typeof(timeout) == 'number' && timeout > 0) {
msg = self.ws.read(timeout)
lastRead = true
}
}
if (msg != null && msg != "") {
if (msg != "pong" && msg != "ping") {
self.processMsg(msg)
}
} else {
break
}
if (lastRead) {
break
}
}
}
return self
}
function NewOKXWebSocketPublic(e, onLogin, onTick) {
var crc32 = function(r) {
for (var a, o = [], c = 0; c < 256; c++) {
a = c;
for (var f = 0; f < 8; f++) a = 1 & a ? 3988292384 ^ a >>> 1 : a >>> 1;
o[c] = a
}
for (var n = -1, t = 0; t < r.length; t++) n = n >>> 8 ^ o[255 & (n ^ r.charCodeAt(t))];
return (-1 ^ n) >>> 0
};
let self = {
e: e,
key: e.GetName() + '/' + e.GetCurrency(),
quoteCurrency: e.GetQuoteCurrency(),
name: e.GetName(),
isFutures: e.GetName().indexOf("Futures_") == 0,
ws: null,
cache: {},
channles: [],
depthCount: 0,
depthConsumed: 0,
lastPing: 0,
depthDic: {},
lastMarket: 0
}
self.wsPrivate = NewOKXWebSocketPrivate(self)
self.processTick = function(event) {
if (typeof(onTick) !== 'function') {
return
}
let ret = onTick(self, event)
if (ret) {
if (ret.newOrders && ret.newOrders.length > 0) {
ret.ctx.wsPrivate.batchOrders(ret.newOrders)
}
if (ret.amendOrders && ret.amendOrders.length > 0) {
ret.ctx.wsPrivate.amendOrders(ret.amendOrders)
}
if (ret.cancelOrders && ret.cancelOrders.length > 0) {
ret.ctx.wsPrivate.cancelOrders(ret.cancelOrders)
}
}
}
self.processMsg = function(msg) {
if (Debug) {
Log("MSG:", msg)
}
let obj = JSON.parse(msg)
if (obj.event == 'subscribe') {
self.channles.push(obj)
} else if (obj.event == 'error') {
throw obj.msg
} else if (obj.event == 'login') {
self.channles = []
if (typeof(onLogin) === 'function') {
onLogin(self.ws, self.e)
}
} else {
self.lastMarket = new Date().getTime();
let instId = obj.arg.instId
let event = {
instId: instId
}
obj.data.forEach(function(item) {
if (obj.arg.channel == 'trades') {
if (typeof(event.trades) === 'undefined') {
event.trades = []
}
event.ts = Number(item.ts)
event.trades.push({
ts: Number(item.ts),
side: item.side,
price: Number(item.px),
qty: Number(item.sz)
})
} else if (obj.arg.channel == 'books5' || obj.arg.channel == 'books' || obj.arg.channel == 'books50-l2-tbt') {
let tsBegin = UnixNano()
let depth = {
asks: [],
bids: []
}
if (obj.arg.channel == 'books5') {
item.asks.forEach(function(pair) {
depth.asks.push({
price: Number(pair[0]),
qty: Number(pair[1])
})
})
item.bids.forEach(function(pair) {
depth.bids.push({
price: Number(pair[0]),
qty: Number(pair[1])
})
})
} else {
let depthDic = self.depthDic[instId]
if (typeof(depthDic) === 'undefined') {
depthDic = {
count: 0,
dic: {
asks: {},
bids: {}
}
}
self.depthDic[instId] = depthDic
}
depthDic.count += 1
for (let k in depthDic.dic) {
if (obj.action == 'snapshot') {
depthDic.dic[k] = {}
}
let mp = depthDic.dic[k]
item[k].forEach(function(book) {
if (book[1] == '0') {
delete mp[book[0]]
} else {
mp[book[0]] = [book[1], book[3], item['ts']]
}
})
}
for (let k in depth) {
let n = k == 'asks' ? 1 : -1
let mp = depthDic.dic[k]
Object.keys(depthDic.dic[k]).sort(function(a, b) {
return n * (Number(a) - Number(b))
}).forEach(function(x) {
// keep string for
depth[k].push({
price: x,
qty: mp[x][0]
})
})
}
if (depthDic.count % 5000 == 0) {
let s = []
for (let i = 0; i < 25; i++) {
['bids', 'asks'].forEach(function(k) {
if (i < depth[k].length) {
s.push(depth[k][i].price + ':' + depth[k][i].qty)
}
})
}
if (crc32(s.join(":")) != Uint32Array.from(Int32Array.of(item.checksum))[0]) {
throw "depth checksum error"
}
}
// convert to number
for (let dir in depth) {
let books = depth[dir]
for (let i = 0; i < books.length; i++) {
books[i].price = Number(books[i].price)
books[i].qty = Number(books[i].qty)
}
}
}
event.depth = depth
event.ts = Number(item.ts)
}
})
if (event.ts) {
self.processTick(event)
}
}
}
self.isReady = function() {
return self.wsPrivate.isReadyDic["account"] && self.wsPrivate.isReadyDic["orders"]
}
self.reset = function() {
if (self.ws) {
self.ws.close()
self.ws = null
}
self.depthDic = {}
self.isReadyDic = {}
}
self.poll = function(timeout) {
let ts = new Date().getTime()
if (self.lastPing == 0) {
self.lastPing = ts
}
if (!self.ws) {
self.ws = Dial("wss://ws.okx.com:8443/ws/v5/public")
if (!self.ws) {
return
}
if (typeof(AccessKey) === 'undefined' || AccessKey.length == 0) {
if (typeof(onLogin) === 'function') {
onLogin(self.ws, self.e)
}
} else {
let tsStr = (new Date().getTime() / 1000).toString()
let authMsg = {
"op": "login",
"args": [{
"apiKey": AccessKey,
"passphrase": Passphrase,
"timestamp": tsStr,
"sign": self.e.HMAC("sha256", "base64", tsStr + "GET" + "/users/self/verify", "{{secretkey}}")
}]
}
self.ws.write(JSON.stringify(authMsg))
}
}
if (!self.ws) {
return
}
if (ts - self.lastPing > 10000) {
self.ws.write("ping")
self.lastPing = ts
}
let lastRead = false
while (true) {
let msg = self.ws.read(-1)
if (msg == "") {
self.reset()
break
}
if (msg == null) {
if (typeof(timeout) == 'number' && timeout > 0) {
msg = self.ws.read(timeout)
lastRead = true
}
}
if (msg != null && msg != "") {
if (msg != "pong") {
self.wsPrivate.poll()
self.processMsg(msg)
self.wsPrivate.poll()
}
} else {
break
}
if (lastRead) {
break
}
}
self.wsPrivate.poll()
}
return self
}
function NewBinanceSocketPrivate(ctx) {
let self = {
ws: null,
isFutures: ctx.isFutures,
isReadyDic: {},
lastPing: 0,
ctx: ctx,
useMargin: UseMargin,
quoteCurrency: ctx.e.GetQuoteCurrency(),
listenKey: null,
listenKeyUpdate: 0,
account: {
cancelPending: {},
orders: {},
ordersCount: {
pending: 0,
buy: 0,
amend: 0,
canceled: 0,
filled: 0,
push: 0,
sell: 0
},
positions: {},
balance: {},
init_balance: {},
balanceUpdate: 0,
positionsUpdate: 0,
ordersUpdate: 0
}
}
if (ctx.isFutures) {
// TODO change to api
let positions = _C(exchange.GetPosition)
positions.forEach(function(pos) {
// avoid self.instId
self.account.positions[self.instId] = {
Amount: pos.Amount * (pos.Type == PD_LONG ? 1 : -1),
Price: pos.Price
}
})
}
let pair = ctx.e.GetCurrency().split('_')
if (self.useMargin && !ctx.isFutures) {
let ret = _C(ctx.e.IO, "api", "GET", "/sapi/v1/margin/account")
ret.userAssets.forEach(function(item) {
if (item.asset == pair[0] || item.asset == pair[1]) {
self.account.balance[item.asset] = {
free: Number(item.free),
locked: Number(item.locked),
borrowed: Number(item.borrowed),
}
self.account.init_balance[item.asset] = {
free: Number(item.free),
locked: Number(item.locked),
borrowed: Number(item.borrowed)
}
}
})
self.tradePath = "/sapi/v1/margin/order"
} else {
let acc = _C(ctx.e.GetAccount)
self.account.balance[pair[1]] = {
free: acc.Balance,
locked: acc.FrozenBalance,
borrowed: 0
}
self.account.balance[pair[0]] = {
free: acc.Stocks,
locked: acc.FrozenStocks,
borrowed: 0
}
self.account.init_balance[pair[1]] = {
free: acc.Balance,
locked: acc.FrozenBalance,
borrowed: 0
}
self.account.init_balance[pair[0]] = {
free: acc.Stocks,
locked: acc.FrozenStocks,
borrowed: 0
}
self.tradePath = "/api/v3/order"
}
Log(ctx.e.GetName(), self.account.init_balance)
self.getOrdersMap = function(instId) {
let mp = self.account.orders[instId]
if (typeof(mp) === 'undefined') {
mp = {}
self.account.orders[instId] = mp
}
return mp
}
self.reset = function() {
if (self.ws) {
self.ws.close()
self.ws = null
}
self.isReadyDic = {}
}
self.send = function(msg) {
if (!self.ws) {
return
}
self.ws.write(JSON.stringify(msg))
}
self.batchOrders = function(orders) {
let args = []
let ts = new Date().getTime()
orders.forEach(function(item) {
args.push({
symbol: item.instId,
side: item.side.toUpperCase(),
newClientOrderId: item.clOrdId,
type: "LIMIT",
price: item.px,
quantity: item.sz,
timeInForce: "GTX",
})
self.getOrdersMap(item.instId)[item.clOrdId] = {
Id: item.clOrdId,
Created: ts,
instId: item.instId,
Price: Number(item.px),
Amount: Number(item.sz),
Type: item.side == "sell" ? ORDER_TYPE_SELL : ORDER_TYPE_BUY,
DealAmount: 0,
Status: ORDER_STATE_PENDING
}
})
let rets = []
if (self.ctx.isFutures) {
for (let i = 0; i < args.length; i += 5) {
let ret = self.ctx.e.IO("api", "POST", "/fapi/v1/batchOrders", "", JSON.stringify({
batchOrders: args.slice(i, i + 5)
}))
if (Debug) {
Log("batchOrders", args, ret)
}
rets.push(ret)
}
} else {
orders.forEach(function(order) {
let ret = self.ctx.e.IO("api", "POST", self.tradePath, "", JSON.stringify(order))
if (Debug) {
Log("batchOrders", args, ret)
}
rets.push(ret)
})
}
return rets
}
self.cancelOrders = function(orders) {
let ts = new Date().getTime()
orders.forEach(function(item) {
if (typeof(self.account.cancelPending[item.clOrdId]) === 'undefined') {
self.account.cancelPending[item.clOrdId] = {
retry: 0,
ts: ts,
data: item
}
}
})
for (let id in self.account.cancelPending) {
let item = self.account.cancelPending[id]
if (ts - item.ts > 10000) {
// recancel
orders.push(item.data)
Log("remove timeout order", item, "#ff0000")
item.retry += 1
item.ts = ts
}
if (item.retry > 10) {
Log("force remove order", item, "#ff0000")
delete self.account.cancelPending[id]
}
}
let mp = {}
orders.forEach(function(item) {
if (typeof(mp[item.instId]) === 'undefined') {
mp[item.instId] = []
}
mp[item.instId].push(item.clOrdId)
delete self.getOrdersMap(item.instId)[item.clOrdId]
})
let rets = []
if (self.ctx.isFutures) {
for (let instId in mp) {
let arr = mp[instId]
for (let i = 0; i < arr.length; i += 5) {
let ids = arr.slice(i, i + 5)
let ret = self.ctx.e.IO("api", "DELETE", "/fapi/v1/batchOrders", "", JSON.stringify({
symbol: instId,
origClientOrderIdList: ids
}))
if (Debug) {
Log("cancelOrders", instId, arr, ret)
}
// cancel success
if (ret && ret.length == ids.length) {
ids.forEach(function(id) {
delete self.account.cancelPending[id]
})
}
rets.push(ret)
}
}
} else {
for (let i = 0; i < orders.length; i += 5) {
let ret = self.ctx.e.IO("api", "DELETE", self.tradePath, "", JSON.stringify(orders[i]))
if (Debug) {
Log("cancelOrders", orders[i], ret)
}
rets.push(ret)
}
}
return rets
}
self.processOrders = function(ts, data, reset) {
if (Debug) {
Log("ORDERS: ", data)
}
if (reset) {
self.account.orders = {}
for (let k in self.account.ordersCount) {
self.account.ordersCount[k] = 0
}
}
let ret = []
data.forEach(function(item) {
let instId = item.s
let dataOrder = {
Id: item.C || item.c,
instId: instId,
Info: item,
Price: Number(item.p),
Amount: Number(item.q),
Update: Number(item.T || item.E || ts),
Created: Number(item.O || item.E || ts),
Type: item.S == "BUY" ? ORDER_TYPE_BUY : ORDER_TYPE_SELL,
DealAmount: Number(item.z)
}
if (typeof(item.ap) !== 'undefined') {
dataOrder.AvgPrice = Number(item.ap)
} else if (Number(item.z) > 0 && typeof(item.Z) !== 'undefined') {
dataOrder.AvgPrice = Number(item.Z) / Number(item.z)
}
if (item.X == "NEW") {
dataOrder.Status = ORDER_STATE_PENDING
self.account.ordersCount.push += 1
self.account.ordersCount.pending += 1
} else if (item.X == "PARTIALLY_FILLED") {
dataOrder.Status = ORDER_STATE_PENDING
} else if (item.X == "FILLED") {
dataOrder.Status = ORDER_STATE_CLOSED
self.account.ordersCount.filled += 1
self.account.ordersCount.pending -= 1
} else if (item.X == "CANCELED" || item.X == "EXPIRED" || item.X == "REJECT" || item.X == "NEW_INSURANCE" || item.X == "NEW_ADL") {
dataOrder.Status = ORDER_STATE_CANCELED
self.account.ordersCount.canceled += 1
self.account.ordersCount.pending -= 1
} else {
throw "unknow order state " + JSON.stringify(item)
}
// remove from cancelPending orders
if (dataOrder.Status != ORDER_STATE_PENDING) {
delete self.account.cancelPending[dataOrder.Id]
if (Debug) {
Log("remove from cancel pending orders", dataOrder)
}
}
let mp = self.getOrdersMap(instId)
let oldOrder = mp[dataOrder.Id]
if ((!oldOrder) || (dataOrder.DealAmount != oldOrder.DealAmount)) {
if (item.X == "PARTIALLY_FILLED" || item.X == "FILLED") {
Log(item.X, item.S, dataOrder)
}
}
let update = false
if (dataOrder.Status == ORDER_STATE_PENDING) {
// 修改订单的命令还没返回就收到了new订单的update导致时间序列错乱
if (oldOrder) {
update = true
Object.assign(oldOrder, dataOrder)
} else {
mp[dataOrder.Id] = dataOrder
update = true
}
if (Debug) {
Log("update order", Boolean(oldOrder), mp[dataOrder.Id], item)
}
} else {
update = true
if (oldOrder) {
// avoid ref
Object.assign(oldOrder, dataOrder)
delete mp[dataOrder.Id]
}
if (Debug) {
Log("order " + item.X, dataOrder)
}
}
if (update) {
ret.push(dataOrder)
}
self.account.ordersUpdate = ts
})
return ret
}
self.processPrivateMsg = function(msg) {
let obj = JSON.parse(msg)
if (Debug) {
Log(obj, "#ff0000")
}
let event = {}
if (obj.e == "ORDER_TRADE_UPDATE") {
event.orders = self.processOrders(obj.E, [obj.o])
event.ts = obj.E
self.isReadyDic["orders"] = true
} else if (obj.e == "ACCOUNT_UPDATE" && obj.a) {
event.ts = obj.E
self.account.balanceUpdate = obj.E
if (obj.a.B) {
obj.a.B.forEach(function(item) {
self.account.balance[item.a] = {
free: Number(item.wb),
locked: 0
}
})
event.balance = self.account.balance
}
if (obj.a.P) {
obj.a.P.forEach(function(item) {
self.account.positions[item.s] = {
Amount: Number(item.pa),
Price: Number(item.ep)
}
self.account.positionsUpdate = obj.E
})
event.positions = self.account.positions
}
self.isReadyDic["account"] = true
} else if (obj.e == "outboundAccountPosition") {
event.ts = obj.E
self.account.balanceUpdate = obj.E
if (obj.B) {
obj.B.forEach(function(item) {
self.account.balance[item.a] = {
free: Number(item.f),
locked: Number(item.l)
}
})
event.balance = self.account.balance
}
self.isReadyDic["account"] = true
} else if (obj.e == "executionReport") {
event.orders = self.processOrders(obj.E, [obj])
event.ts = obj.E
self.isReadyDic["orders"] = true
} else {
//Log("RECV", "<" + msg + ">")
}
if (event.ts) {
self.ctx.processTick(event)
}
}
self.poll = function(timeout) {
let ts = new Date().getTime()
if (self.lastPing == 0) {
self.lastPing = ts
}
if (!self.listenKey || ts - self.listenKeyUpdate > 60000 * 30) {
let ret = self.ctx.e.IO("api", "POST", self.ctx.isFutures ? "/fapi/v1/listenKey" : (self.useMargin ? "/sapi/v1/userDataStream" : "/api/v3/userDataStream"))
if (ret && ret.listenKey) {
self.listenKey = ret.listenKey
self.listenKeyUpdate = ts
//Log("ListenKey Update", ret)
}
}
if (!self.listenKey) {
return
}
if (self.ws == null) {
let base = (self.ctx.isFutures ? 'wss://fstream.binance.com' : 'wss://stream.binance.com:9443') + "/ws"
Log("Dial: ", base + "/*****")
self.ws = Dial(base + "/" + self.listenKey)
}
if (!self.ws) {
return;
}
let lastRead = false
while (true) {
let msg = self.ws.read(-1)
if (msg == "") {
Log("Binance private websocket reset")
self.reset()
break
}
if (msg == null) {
if (typeof(timeout) == 'number' && timeout > 0) {
msg = self.ws.read(timeout)
lastRead = true
}
}
if (msg != null && msg != "") {
//Log("BP", msg)
if (msg == "ping") {
self.ws.write("pong")
self.lastPing = ts
} else if (msg != "pong") {
self.processPrivateMsg(msg)
}
} else {
break
}
if (lastRead) {
break
}
}
}
return self
}
function NewBinanceWebSocketPublic(e, onLogin, onTick) {
let self = {
e: e,
key: e.GetName() + '/' + e.GetCurrency(),
quoteCurrency: e.GetQuoteCurrency(),
name: e.GetName(),
isFutures: e.GetName().indexOf("Futures_") == 0,
ws: null,
cache: {},
depthCount: 0,
depthConsumed: 0,
lastPing: 0,
isReadyDic: {}
}
self.base = self.isFutures ? "wss://fstream.binance.com/ws" : "wss://stream.binance.com/ws"
self.wsPrivate = NewBinanceSocketPrivate(self)
self.reset = function() {
if (self.ws) {
self.ws.close()
self.ws = null
}
self.isReadyDic = {}
}
self.processTick = function(event) {
if (typeof(onTick) !== 'function') {
return
}
let ret = onTick(self, event)
if (ret) {
if (ret.amendOrders && ret.amendOrders.length > 0) {
ret.ctx.wsPrivate.amendOrders(ret.amendOrders)
}
if (ret.cancelOrders && ret.cancelOrders.length > 0) {
ret.ctx.wsPrivate.cancelOrders(ret.cancelOrders)
}
if (ret.newOrders && ret.newOrders.length > 0) {
ret.ctx.wsPrivate.batchOrders(ret.newOrders)
}
}
}
self.processMsg = function(msg) {
if (Debug) {
Log("MSG:", msg)
}
let obj = JSON.parse(msg)
if (obj.error) {
Log(obj)
throw obj.error.msg
}
let event = {
ts: obj.E,
instId: obj.s
}
if (obj.e == "depthUpdate") {
let depth = {
asks: [],
bids: []
}
obj.b.forEach(function(item) {
depth.bids.push({
price: Number(item[0]),
qty: Number(item[1])
})
})
obj.a.forEach(function(item) {
depth.asks.push({
price: Number(item[0]),
qty: Number(item[1])
})
})
event.depth = depth
//self.depthTime = obj.data.E
} else if (obj.e == 'aggTrade') {
event.ts = obj.E
event.trades = [{
price: Number(obj.p),
qty: Number(obj.q),
ts: obj.T,
side: obj.m ? "sell" : "buy"
}]
} else if (typeof(obj.asks) !== 'undefined') {
let depth = {
asks: [],
bids: []
}
obj.bids.forEach(function(item) {
depth.bids.push({
price: Number(item[0]),
qty: Number(item[1])
})
})
obj.asks.forEach(function(item) {
depth.asks.push({
price: Number(item[0]),
qty: Number(item[1])
})
})
event.ts = obj.E || new Date().getTime()
event.depth = depth
} else {
Log(">>>", msg)
return
}
self.processTick(event)
}
self.isReady = function() {
return self.wsPrivate.ws != null
}
self.poll = function(timeout) {
let ts = new Date().getTime()
if (!self.ws) {
Log("Dial: ", self.base)
self.ws = Dial(self.base)
if (self.ws) {
onLogin(self.ws, self.e)
}
}
if (!self.ws) {
return
}
let lastRead = false
while (true) {
let msg = self.ws.read(-1)
if (msg == "") {
if (Debug) {
Log("DEBUG> RESET websocket", self.base, "#ff0000")
}
self.reset()
break
}
if (msg == null) {
if (typeof(timeout) == 'number' && timeout > 0) {
msg = self.ws.read(timeout)
lastRead = true
}
}
if (msg != null && msg != "") {
if (msg == "ping") {
self.ws.write("pong")
self.lastPing = ts
} else if (msg == "pong") {
// ignore
} else {
self.wsPrivate.poll()
self.processMsg(msg)
self.wsPrivate.poll()
}
} else {
break
}
if (lastRead) {
break
}
}
self.wsPrivate.poll()
}
return self
}
if (typeof(__threadId) == 'function' && __threadId() != 0) {
// thread register
this.NewWSS = function(e, onWSSLogin, onWSSTick) {
let pfn = null
if (e.GetName().indexOf("Binance") != -1) {
pfn = NewBinanceWebSocketPublic
Log("New Binance websocket", e.GetName(), e.GetCurrency())
} else {
pfn = NewOKXWebSocketPublic
Log("New OKX Websocket", e.GetName(), e.GetCurrency())
}
return pfn(e, onWSSLogin, onWSSTick)
}
return
}
let pfn = null
if (e.GetName().indexOf("Binance") != -1) {
pfn = NewBinanceWebSocketPublic
Log("New Binance websocket", e.GetName(), e.GetCurrency())
} else {
pfn = NewOKXWebSocketPublic
Log("New OKX Websocket", e.GetName(), e.GetCurrency())
}
return pfn(e, onWSSLogin, onWSSTick)
}
function onTick(ctx, event) {
if (event.depth) {
Log("depth update", event)
}
if (event.trades) {
Log("trades received", event)
}
if (event.balance || event.positions) {
Log("account update", event)
}
if (event.orders) {
Log("private orders update", event)
}
// include orders, positions, balance for latest
Log("account", ctx.wsPrivate.account)
// for test only
if (false) {
return {
amendOrders: [{
instId: event.instId,
clOrdId: "xxxx*****",
cxlOnFail: true,
newSz: "2",
}],
newOrders: [{
instId: event.instId,
clOrdId: UUID(),
side: "sell",
tdMode: "cross",
ordType: "post_only",
px: event.depth.asks[0].price.toFixed(4),
sz: "1",
}, {
instId: event.instId,
clOrdId: UUID(),
side: "sell",
tdMode: "cross",
ordType: "post_only",
px: event.depth.asks[0].price.toFixed(4),
sz: "1",
}],
cancelOrders: [{
instId: order.instId,
clOrdId: order.Id
}]
}
}
}
function main() {
let instId = exchange.SetContractType("swap").InstrumentID
let ctx = $.NewWSS(exchange, function(ws) {
let msg = null
if (exchange.GetName() === 'Futures_OKCoin') {
msg = {
op: "subscribe",
args: [{
channel: "books5",
instId: instId,
}, {
channel: "trades",
instId: instId,
}]
}
} else {
let symbol = exchange.GetCurrency().replace('_', '').toLowerCase()
msg = {
method: "SUBSCRIBE",
params: [symbol + "@aggTrade", symbol + "@depth20@100ms"],
id: 1,
}
}
ws.write(JSON.stringify(msg))
Log("subscribe", msg, "channel")
LogStatus("Ready")
}, onTick, false)
while (true) {
ctx.poll()
EventLoop(1000)
}
}