diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2095605 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +tplog_* +db/ +logs/ +data/ +sym_px \ No newline at end of file diff --git a/tickplant/bootstrap.zsh b/tickplant/bootstrap.zsh new file mode 100755 index 0000000..29a7424 --- /dev/null +++ b/tickplant/bootstrap.zsh @@ -0,0 +1,50 @@ +#!/bin/zsh +# $ ./bootstrap.zsh [run|stop] +# This script is used to bootstrap the tickplant service. It is called by the +# first start the ticker process, then the feedhandler process, then the hdb etc +# The script also stops the processes if they are running. + +#find if another process is running on 5010 and kill it if it is +echo "Stopping ticker|feed|hdb processes" +lsof -i :5010 | grep q | awk '{print $2}' | xargs kill -SIGTERM +lsof -i :5011 | grep q | awk '{print $2}' | xargs kill -SIGTERM +lsof -i :5012 | grep q | awk '{print $2}' | xargs kill -SIGTERM + + + +# stop the processes if running in mode stop +if [[ $1 == "stop" ]]; then + echo "Stopped tickplant processes" + exit 0 +fi + +if [[ $1 == "clean" ]]; then + echo "Cleaned logs" + rm -v 5010 + rm -v 5012 + rm -rfv data/* + rm sym_px + exit 0 +fi +set -e # exit on error +alias q="QHOME=~/shadow/q rlwrap -H /var/tmp/hist${PPID} /Users/emelis/shadow/q/m64/q" +TICK_LOG=logs/tick.log +q tick.q -t 1000 -c 31 317 2>&1 > ${TICK_LOG} & +TICK="${!}" +echo "TICKER PID: ${TICK} " + +#find if another instance of feedhandler is running and kill it if it is +ps -ef | grep feedhandler | grep -v grep | awk '{print $2}' | xargs kill -SIGTERM +q feedhandler/feedhandler.q :5010 -t 1000 -c 31 317 2>&1 > logs/feedhandler.log & +FH="${!}" +echo "FH PID: ${FH}" + +q -p 5012 -c 31 317 > logs/hdb.log & +HDB="${!}" +echo "HDB PID: ${HDB}" + +ps -ef | grep orderbook | grep -v grep | awk '{print $2}' | xargs kill -SIGTERM +q orderbook/orderbook.q :5010 -t 1 -c 31 317 2>&1 > logs/orderbook.log & +ORDERBOOK="${!}" +echo "ORDERBOOK PID: ${ORDERBOOK}" + diff --git a/tickplant/cep/cep.q b/tickplant/cep/cep.q new file mode 100644 index 0000000..45d4d22 --- /dev/null +++ b/tickplant/cep/cep.q @@ -0,0 +1,45 @@ +/q cep/cep.q SRC :5010 + +system"l tick/r.q" +system"l common/conn.q" +system"l cep/lib.q" + +.u.conn:{hopen `$":",x}each enlist[`tick]#.u.x; + +// uend:{ +// if[null x;'daycantbenull]; +// // t:tables`.; +// t:`trade`order; +// t@:where `g=attr each t@\:`sym; +// .Q.hdpf[.u.x `hdb;HDBROOT;x;`sym]; +// @[;`sym;`g#] each t; +// }; + +// k).Q.hdpf:{[h;d;p;f](@[`.;;0#]dpft[d;p;f]@)'t@>(#.:)'t:`trade`order;if[h:@[hopen;h;0];h"\\l .";>h]} +// .u.end:{ +// if[null x;'daycantbenull]; +// .cep.end x; +// uend x; +// } + +.cep.start:{ + .log.logInfo "kdb+tick CEP date:",string[.z.P]," version:",string[.z.K],"_",string .z.k; + .cep.replay::1b; + .u.rep .(.u.conn`tick)"(.u.sub[`;`];`.u `i`L)"; + .cep.replay::0b; + } + +.cep.lastTs:.z.P; + +.cep.ts:{ + // if[.z.P-.cep.lastTs>=00:01:00.000; ohlcvv_1m:: calcOHLCVV_1m[]]; + // mid_1m::calcMID_1m[]; + } + +.z.ts:{ + .log.logInfo"Running CEP ts"; + .u.ts[]; + .cep.ts[]; + } + +.cep.start[] \ No newline at end of file diff --git a/tickplant/cep/lib.q b/tickplant/cep/lib.q new file mode 100644 index 0000000..0745433 --- /dev/null +++ b/tickplant/cep/lib.q @@ -0,0 +1,63 @@ +calcOHLCVV_1m:{ + select + open: first price, + high: max price, + low: min price, + close: last price, + volume: sum size, + vwap: size wavg price + by sym, 1 xbar time.minute + from trade + } + +calcOHLCVV_1d:{ + select + open: first price, + high: max price, + low: min price, + close: last price, + volume: sum size, + vwap: size wavg price + by sym + from trade + } + +// For sampling purposes only - not a realistic calculation +calcTickMID:{select mid_price:avg price by exch, sym, minute: 1 xbar time.minute from order} + +calcVWAP:{[s;e;r] + $[r=1; select sym, exch, second, vwap,res:r from vwap_1s where sym=s,exch=e; + select vwap: sum[wsp]%sum ws, sum num_trade by sym,exch,r xbar second from vwap_1s where sym=s,exch=e] + }; + +last_ts:.z.P; + + + +// .cep.end:{.cep.ts[]; ohlcvv_1d::calcOHLCVV_1d[];} + +upd:{ + .[insert;(x;y);{.log.logError"Insert failed for table=",string[y]," with error=",x}[;x]]; + if[`trade=x; + if[not .cep.replay; lastTrade,:y]; + // having this separate table is questionable - native wavg performs well compared to running sum x*y%sum y + // but it feels like there is a scale point at which it will be faster to use the running sum when calculating + // multiple vwap horizons instead of recalculating the wavg for each horizon + // We shall see... + // leaving vwap here to compare results with the native wavg + // if[99h=type y; y:enlist y]; + // vwap_1s+: select wsp:size wsum price, ws:sum size, vwap:size wavg price, num_trade:count i by exch,sym,time.second from trade + // where ([]time.second;sym;exch) in select time.second,sym,exch from y + ]; + if[`order=x; if[not .cep.replay; lastOrder,:y]]; + }; + +// MACD +ema12:ema[2%13;] +ema26:ema[2%17;] +macd: {ema12[x] - ema26 x} +signal:{ema[2%10;macd x]} + +calcMACD:{[s;e] + select minute, exch, ema12 close, ema26 close, macd close, signal macd close by sym from calcOHLCVV_1m[] + } \ No newline at end of file diff --git a/tickplant/cep/schema.q b/tickplant/cep/schema.q new file mode 100644 index 0000000..3a801dd --- /dev/null +++ b/tickplant/cep/schema.q @@ -0,0 +1,3 @@ +HDB_PERSIST:`trade`order`vwap_1s; + +vwap_1s:update `g#sym from ([exch:`$();sym:`$();second:`second$()];wsp:`float$();ws:`float$();vwap:`float$();num_trade:`long$()); diff --git a/tickplant/common/conn.q b/tickplant/common/conn.q new file mode 100644 index 0000000..7d147d9 --- /dev/null +++ b/tickplant/common/conn.q @@ -0,0 +1,15 @@ +/ get the ticker plant and history ports, defaults are 5010,5012 +.u.x:`tick`hdb!.z.x,(count .z.x)_(":5010";":5012"); + +.z.pc:{show "Process at handle ",string[p:.u.conn ? x]," disconnected"; .u.pc p} + +.u.pc:{.u.conn[x]:0Ni; show .u.conn} + +.u.ts:{ + if[ null .u.conn`tick; + show "Retrying tickerplant connection ..."; + h:@[hopen;`$":",.u.x`tick;{show "Cannot establish tickerplant connection with ",x; 0Ni}]; + .u.conn[`tick]: h; + if[ not null .u.conn`tick; .u.rep .(.u.conn`tick)"(.u.sub[`;`];`.u `i`L)"]; + ]; + } \ No newline at end of file diff --git a/tickplant/common/log.q b/tickplant/common/log.q new file mode 100644 index 0000000..03a4ff9 --- /dev/null +++ b/tickplant/common/log.q @@ -0,0 +1,27 @@ +if[`log in key`.;:(`log_already_loaded)]; +\d .log +LEVELS:`info`warning`error`debug +// .log.test[`something]: {(levels ? `info) >= levels ? `warning} +// Use +// info when you want to log high level state and flow information +// warning when you want to log messages that indicates non-ideal state or potential problems +// error when you want to log meesages that indicate a non-recoverable error +// debug when you want to log a debug message +level: LEVELS ? `debug +setLogLevel:{[l] .log.level::LEVELS ? l}; +canLog:{.log.level <= .log.LEVELS ?x} + +logmsg:{[l;msg] + if[not canLog l; :(::)]; + if[11h = type msg; msg:string msg]; + 0N!string[.z.P]," ",upper[string l]," ",msg; + } + +logInfo:logmsg[`info] +logWarning:logmsg[`warning] +logError:logmsg[`error] +logDebug:logmsg[`debug] + +// TODO be nice to have both stdout/stderr and file logging +// and also only print to stdout if the level is lower than set up +\d . \ No newline at end of file diff --git a/tickplant/common/schema.q b/tickplant/common/schema.q new file mode 100644 index 0000000..9f5229d --- /dev/null +++ b/tickplant/common/schema.q @@ -0,0 +1,7 @@ +trade:([]time:`timestamp$();exch:`symbol$();sym:`symbol$();price:`float$();size:`float$();side:`symbol$(); tradeid:`guid$()); + +quote:([]time:`timestamp$();exch:`symbol$();sym:`symbol$();bprice:`float$();bsize:`float$();asize:`float$();aprice:`float$()); + +order:([]time:`timestamp$();exch:`symbol$();sym:`symbol$();price:`float$();size:`float$();side:`symbol$();orderid:`guid$()); + +orderbookl2:([]time:`timestamp$(); exch:`symbol$(); sym:`symbol$(); ind:`int$(); btime:`timestamp$();btot_orders:`int$();bsz:`float$();bpx:`float$();apx:`float$();asz:`float$();atot_orders:`int$();atime:`timestamp$()); \ No newline at end of file diff --git a/tickplant/common/sp500.txt b/tickplant/common/sp500.txt new file mode 100644 index 0000000..f66cc11 --- /dev/null +++ b/tickplant/common/sp500.txt @@ -0,0 +1,500 @@ +A.N +AA.N +AAPL.O +ABC.N +ABI.N +ABK.N +ABS.N +ABT.N +ACE.N +ACS.N +ACV.N +ADBE.O +ADCT.O +ADI.N +ADM.N +ADP.N +ADSK.O +AEE.N +AEP.N +AES.N +AET.N +AFL.N +AGN.N +AHC.N +AIG.N +AIV.N +ALL.N +ALTR.O +AMAT.O +AMCC.O +AMD.N +AMGN.O +AMP.N +AMZN.O +AN.N +ANDW.O +AOC.N +APA.N +APC.N +APCC.O +APD.N +APOL.O +ASD.N +ASH.N +ASN.N +ASO.N +AT.N +ATI.N +AV.N +AVP.N +AVY.N +AW.N +AXP.N +AYE.N +AZO.N +BA.N +BAC.N +BAX.N +BBBY.O +BBT.N +BBY.N +BC.N +BCR.N +BDK.N +BDX.N +BEN.N +BFb.N +BHI.N +BIIB.O +BJS.N +BK.N +BLI.N +BLL.N +BLS.N +BMC.N +BMET.O +BMS.N +BMY.N +BNI.N +BOL.N +BR.N +BRCM.O +BSC.N +BSX.N +BUD.N +C.N +CA.N +CAG.N +CAH.N +CAT.N +CB.N +CBE.N +CBSS.O +CC.N +CCE.N +CCL.N +CCU.N +CD.N +CEG.N +CFC.N +CHIR.O +CI.N +CIEN.O +CIN.N +CINF.O +CIT.N +CL.N +CLX.N +CMA.N +CMCSA.O +CMI.N +CMS.N +CMVT.O +CMX.N +CNP.N +COF.N +COH.N +COL.N +COP.N +COST.O +CPB.N +CPN.N +CPWR.O +CSC.N +CSCO.O +CSX.N +CTAS.O +CTB.N +CTL.N +CTX.N +CTXS.O +CVG.N +CVH.N +CVS.N +CVX.N +CZN.N +D.N +DCN.N +DD.N +DDS.N +DE.N +DELL.O +DG.N +DGX.N +DHI.N +DHR.N +DIS.N +DJ.N +DOV.N +DOW.N +DRI.N +DTE.N +DUK.N +DVN.N +DYN.N +EBAY.O +EC.N +ECL.N +ED.N +EDS.N +EFX.N +EIX.N +EK.N +EMC.N +EMN.N +EMR.N +EOG.N +EOP.N +EP.N +EQR.N +ERTS.O +ESRX.O +ET.N +ETN.N +ETR.N +EXC.N +F.N +FCX.N +FD.N +FDC.N +FDO.N +FDX.N +FE.N +FHN.N +FII.N +FISV.O +FITB.O +FLR.N +FNM.N +FO.N +FPL.N +FRE.N +FRX.N +FSH.N +FSLb.N +GAS.N +GCI.N +GD.N +GDT.N +GDW.N +GE.N +GENZ.O +GILD.O +GIS.N +GLW.N +GM.N +GP.N +GPC.N +GPS.N +GR.N +GS.N +GT.N +GTW.N +GWW.N +HAL.N +HAS.N +HBAN.O +HCA.N +HCR.N +HD.N +HDI.N +HET.N +HIG.N +HLT.N +HMA.N +HNZ.N +HON.N +HOT.N +HPC.N +HPQ.N +HRB.N +HSP.N +HSY.N +HUM.N +IBM.N +IFF.N +IGT.N +INTC.O +INTU.O +IP.N +IPG.N +IR.N +ITT.N +ITW.N +JBL.N +JCI.N +JCP.N +JDSU.O +JNJ.N +JNS.N +JNY.N +JP.N +JPM.N +JWN.N +K.N +KBH.N +KEY.N +KG.N +KLAC.O +KMB.N +KMG.N +KMI.N +KO.N +KR.N +KRB.N +KRI.N +KSE.N +KSS.N +LEG.N +LEH.N +LEN.N +LH.N +LIZ.N +LLL.N +LLTC.O +LLY.N +LMT.N +LNC.N +LOW.N +LPX.N +LSI.N +LTD.N +LTR.N +LU.N +LUV.N +LXK.N +MAR.N +MAS.N +MAT.N +MBI.N +MCD.N +MCK.N +MCO.N +MDP.N +MDT.N +MEDI.O +MEL.N +MER.N +MERQE.O +MET.N +MHP.N +MHS.N +MI.N +MIL.N +MKC.N +MMC.N +MMM.N +MNST.O +MO.N +MOLX.O +MON.N +MOT.N +MRK.N +MRO.N +MSFT.O +MTB.N +MTG.N +MU.N +MUR.N +MWD.N +MWV.N +MXIM.O +MYG.N +MYL.N +NAV.N +NBR.N +NCC.N +NCR.N +NE.N +NEM.N +NFB.N +NI.N +NKE.N +NOC.N +NOV.N +NOVL.O +NSC.N +NSM.N +NTAP.O +NTRS.O +NUE.N +NVDA.O +NVLS.O +NWL.N +NWSa.N +NYT.N +ODP.N +OMC.N +OMX.N +ORCL.O +OXY.N +PAYX.O +PBG.N +PBI.N +PCAR.O +PCG.N +PCL.N +PD.N +PDCO.O +PEG.N +PEP.N +PFE.N +PFG.N +PG.N +PGL.N +PGN.N +PGR.N +PH.N +PHM.N +PKI.N +PLD.N +PLL.N +PMCS.O +PMTC.O +PNC.N +PNW.N +PPG.N +PPL.N +PRU.N +PSA.N +PTV.N +PX.N +Q.N +QCOM.O +QLGC.O +R.N +RAI.N +RBK.N +RDC.N +RF.N +RHI.N +RIG.N +ROH.N +ROK.N +RRD.N +RSH.N +RTN.N +RX.N +S.N +SAFC.O +SANM.O +SBL.N +SBUX.O +SCH.N +SEBL.O +SEE.N +SFA.N +SGP.N +SHLD.O +SHW.N +SIAL.O +SLB.N +SLE.N +SLM.N +SLR.N +SNA.N +SNV.N +SO.N +SOV.N +SPG.N +SPLS.O +SRE.N +STA.N +STI.N +STJ.N +STT.N +STZ.N +SUN.N +SUNW.O +SVU.N +SWK.N +SWY.N +SYK.N +SYMC.O +SYY.N +T.N +TAP.N +TE.N +TEK.N +TER.N +TGT.N +THC.N +TIF.N +TIN.N +TJX.N +TLAB.O +TMK.N +TMO.N +TRB.N +TROW.O +TSG.N +TSN.N +TWX.N +TXN.N +TXT.N +TXU.N +TYC.N +UIS.N +UNH.N +UNM.N +UNP.N +UPS.N +USB.N +UST.N +UTX.N +UVN.N +VC.N +VFC.N +VIAb.N +VLO.N +VMC.N +VNO.N +VZ.N +WAG.N +WAT.N +WB.N +WEN.N +WFC.N +WFT.N +WHR.N +WLP.N +WM.N +WMB.N +WMI.N +WMT.N +WPI.N +WWY.N +WY.N +WYE.N +X.N +XEL.N +XL.N +XLNX.O +XOM.N +XRX.N +XTO.N +YHOO.O +YUM.N +ZION.O +ZMH.N diff --git a/tickplant/feedhandler/feedhandler.q b/tickplant/feedhandler/feedhandler.q new file mode 100644 index 0000000..153940e --- /dev/null +++ b/tickplant/feedhandler/feedhandler.q @@ -0,0 +1,45 @@ +// q feedhandler/feedhandler.q :5010 -c 13 317 -t 100 +// This process simulates a number of orders coming in from the market +// and sends them to the TICKER process +system"l tick/r.q" +system"l common/conn.q" +system"l feedhandler/lib.q" + +.u.end:{'implementme} + +.u.conn:{hopen `$":",x}each enlist[`tick]#.u.x; + +upd:{'implementme} + +SIM_ORDERS:100 + +.fh.i:0; + +.fh.genOrders:{[n;start] + sims:n?0!sym_px; + sds:n?sides; + // Generating integers then dividing to get floats of set precision + // also forcing BUY prices to be generally smaller than SELL prices (can still be crossed) + px:abs sims[`px]+(-1 1)[sds]*floor[sims`px]%100; + sz:0.01|(n?100)%sims`px; + batch:(n#start;n?exchs;sims`sym;px;sz;sds;neg[n]?0Ng); + batch + } + +.fh.ts:{ + n:SIM_ORDERS; + batch:.fh.genOrders[n;.z.p]; + .fh.i+:SIM_ORDERS; + // TODO trap errors like ticker being down + .u.conn[`tick](`.u.upd;`order;batch); + // Simulating market movements + `sym_px set select px:med price,sz:med size by sym from flip cols[order]!batch; + }; + +.z.exit:{`:sym_px set sym_px;} + +.z.ts:{ + .log.logInfo"Running Fedhandler ts"; + .u.ts[]; + .fh.ts[]; + } \ No newline at end of file diff --git a/tickplant/feedhandler/lib.q b/tickplant/feedhandler/lib.q new file mode 100644 index 0000000..9624070 --- /dev/null +++ b/tickplant/feedhandler/lib.q @@ -0,0 +1,9 @@ +exchs:enlist`MYX; + +sides:`u?`B`S; + +syms:1#`$'read0`:common/sp500.txt; + +// Carry price movements across restarts +sym_px:@[get;`:sym_px;{px:(count[syms]?1+100);([sym:syms]px:px;sz:px%10)}]; +`:sym_px set sym_px; \ No newline at end of file diff --git a/tickplant/orderbook/lib.q b/tickplant/orderbook/lib.q new file mode 100644 index 0000000..885b857 --- /dev/null +++ b/tickplant/orderbook/lib.q @@ -0,0 +1,91 @@ +// Stateful resumes from last known state +.orderbook.INSTANCE_PATH:`$":data/orderbook_",.orderbook.INSTANCE,"/" +// NIT - this should be refactored to minimize race conditions between the book timestamp and when it snapshots +// possibly do not snapshot a separate variable +.orderbook.writeSnapshot:{(` sv .orderbook.INSTANCE_PATH,x)set get x;`.orderbook.lastBook set max order_ask[`time],order_bid`time} +.orderbook.readSnapshot:{x set @[get;.orderbook.INSTANCE_PATH,x;{.orderbook[x]}[x]]} + +.orderbook.toSnapshot:{system"x .z.zd";.orderbook.writeSnapshot each `order_ask`order_bid} +.orderbook.fromSnapshot:{.orderbook.readSnapshot each -2?`order_ask`order_bid; `.orderbook.lastBook set max order_ask[`time],order_bid`time} + + +.orderbook.upd:{[t;d] + if[not t=`order; :(`nop)]; + // NIT - could skip messages if for some reason in the same batch there is message with the time before the last book + // but never received from the book. TBC if this would happen in practice + d:update oid:neg[count d]?0Ng,rcvtime:.z.p from d; + `order_ask insert (d where d[`side]=`S); + `order_bid insert (d where d[`side]=`B); + if[00:02:00 < .z.p-.orderbook.lastMatch; .log.logWarning "Nothing matched in over 2 minutes, but data incoming. Is timer on?"]; + } + +// match `sym`exch`time`price`size`side`orderid!(`A.N;`NYSE;.z.P;10;1;`S;"G"$"5ae7962d-49f2-404d-5aec-f7c8abbae288") +// match tries to match an incoming order against the order book +// it will return a trade if the order is fully or partially filled +// it will return a null if there are no fills +// it can either allow for sweeping or not +.orderbook.match:{[x] + // find all orders that can fill x + as:x[`side]; + // this can be refactored + rside:$[as=`S;`order_bid;`order_ask]; + aside:$[as=`S;`order_ask;`order_bid]; + o:`price xdesc $[`S=as; + select from rside where sym=x`sym, exch=x`exch, price>=x`price; + select from rside where sym=x`sym, exch=x`exch, price<=x`price]; + + if[0=count o; :(0Np)]; + + // if[10; + // If aggressor fully filled, and resting order partially filled, send Execution report to both, remove the aggressor and update the resting order + delete from aside where orderid=x`orderid; + trade[`size]:x`size; + update size:qty from rside where orderid=ro`orderid; + ]; + `executions insert trade; + .z.p + } + + +// `time`sym`exch`price`size`side`tradeid`roid`aoid!(.z.p;x`sym;x`exch;x`price;x`size;x`side;last -1?0Ng;x`orderid;last -1?0Ng) +// whatever approach used here, will an amendat do best if used properly? +.orderbook.reportTrade:{ + .log.logInfo"Sending Trade Report for trade=",.Q.s1 x; + .u.conn[`tick](`.u.upd;`trade;x); + } + + +bypx:{select max time,sum size, tot_orders:count i by sym,exch,price from x} +bypx_norm:{ + f:idesc; s:"b"; if[x~`order_ask;f:iasc;s:"a"]; + r:![bypx x;();{x!x}`sym`exch;enlist[`ind]!enlist(f;`price)]; + r:(`sym`exch`ind,`$'s,/:("px";"time";"sz";"tot_orders")) xcol `sym`exch`ind xkey r + } + + +.orderbook.buildl2:{ + r:`sym`exch`bpx xdesc bypx_norm[`order_bid] uj bypx_norm`order_ask; + `time`exch`sym xcols update time:.z.p from .orderbook.l2cols xcols 0!r + } diff --git a/tickplant/orderbook/orderbook.q b/tickplant/orderbook/orderbook.q new file mode 100644 index 0000000..34bcc7e --- /dev/null +++ b/tickplant/orderbook/orderbook.q @@ -0,0 +1,78 @@ +// q orderbook/orderbook.q :5010 -c 13 317 -t 100 -instance one +// This process simulates a bookbuilder/matching engine that +// receives orders from the TICKER process and matches them against the order book +// sending execution reports back to the TICKER process +// It is meant to be run in conjunction with the TICKER process, and has ZERO +// state maintenance across restarts, nor recovery from crashes.. YET + +system"l tick/r.q" +.log.setLogLevel`warning; + + +.orderbook.INSTANCE: "one" +system"l orderbook/lib.q" +system"l orderbook/schema.q" + +system"l common/conn.q" +.u.conn:{hopen `$":",x}each enlist[`tick]#.u.x; + + +.u.end:{'implementme} +.u.rep:{(.[;();:;].)x;if[null first y;:()];-11!y;}; + +// TODO could refactor this? +upd:{ + if[not x=`order; :(`nop)]; + d:y; + if[.orderbook.replay; + d:$[1=count y;enlist;flip] cols[x]!y; + if[.orderbook.lastBook>max d`time; :(`nop)] + ]; + .orderbook.upd[x;d]; + if[not .orderbook.replay; .orderbook.toSnapshot[]]; + } +// When on, an aggressor order will sweep the book until it's fully filled or the book is empty +// When off, an aggressor order will only match against the top of the book +.orderbook.allowSweep:1b; +// Last time there was an execution +.orderbook.lastMatch:0Np; +// Last time .z.ts ran +.orderbook.lastTs:0Np; + + +// Should there be logic to snapshot trades that have been reported already? +// Maybe can change tradeid to be an hash of the trade itself to use as a key +// and keep state for it +.orderbook.start:{ + .log.logInfo "kdb+tick ORDERBOOK date:",string[.z.P]," version:",string[.z.K],"_",string .z.k; + .orderbook.fromSnapshot[]; + .orderbook.replay:1b; + .log.logDebug "Replaying from tpLogs"; + r:(.u.conn`tick)"(.u.sub[`order;`];`.u `i`L)"; + .u.rep . r; + .orderbook.replay:0b; + .log.logDebug "Replaying Done"; + } + +.z.exit:{.log.logInfo"Running OrderBook zExit";.z.ts[];} +// should add a snapshot and replay from snapshot onward + +.z.ts:{ + .log.logInfo"Running OrderBook ts"; + .u.ts[]; + .orderbook.lastTs:.z.p; + // nit - find a better way to do this + offers: 0!select by sym,exch from order_ask where price=(min;price)fby([]sym;exch); + bids: 0!select by sym,exch from order_bid where price=(max;price)fby([]sym;exch); + // This simulations assumes that a crossed book will uncross the next tick by way of trading the top of the book + tomatch: offers,bids; + if[0=count tomatch;:(`nop)]; + .orderbook.lastMatch:max .orderbook.match each tomatch; + // Build after matches, otherwise it'll look like crossed books always + .u.conn[`tick](`.u.upd; `orderbookl2; .orderbook.buildl2[]); // send this to subs? + toReport:select from executions where time>=.orderbook.lastTs; + if[count toReport; .orderbook.reportTrade delete roid,aoid from toReport]; + if[00:02:00 < .z.p-.orderbook.lastMatch; .log.logWarning "Nothing matched in over 2 minutes, is everything okay?"]; + } + +.orderbook.start[] \ No newline at end of file diff --git a/tickplant/orderbook/schema.q b/tickplant/orderbook/schema.q new file mode 100644 index 0000000..3d55df5 --- /dev/null +++ b/tickplant/orderbook/schema.q @@ -0,0 +1,8 @@ +// side bid - aggressing, means selling +.orderbook.order_bid:update `g#sym from ([]time:`timestamp$(); exch:`symbol$(); sym:`symbol$(); price:`float$(); size:`float$();side:`symbol$(); orderid:`guid$(); rcvtime:`timestamp$(); oid:`guid$()); +// side offer - aggressing means buying +.orderbook.order_ask: .orderbook.order_bid + +.orderbook.l2cols: `sym`exch`ind`btime`btot_orders`bsz`bpx`apx`asz`atot_orders`atime; + +executions:([]time:`timestamp$();exch:`symbol$();sym:`symbol$();price:`float$();size:`float$();side:`symbol$(); tradeid:`guid$(); roid:`guid$(); aoid:`guid$()); diff --git a/tickplant/orderbook/test.q b/tickplant/orderbook/test.q new file mode 100644 index 0000000..a57e1e6 --- /dev/null +++ b/tickplant/orderbook/test.q @@ -0,0 +1,4 @@ +// unit test order book matching logic +// generate sample orders table +// cross the spread +// verify fill diff --git a/tickplant/tick.q b/tickplant/tick.q new file mode 100644 index 0000000..727e110 --- /dev/null +++ b/tickplant/tick.q @@ -0,0 +1,89 @@ +/ q tick.q sym . -p 5001 foo 2>&1 & +/2014.03.12 remove license check +/2013.09.05 warn on corrupt log +/2013.08.14 allow when -u is set +/2012.11.09 use timestamp type rather than time. -19h/"t"/.z.Z -> -16h/"n"/.z.P +/2011.02.10 i->i,j to avoid duplicate data if subscription whilst data in buffer +/2009.07.30 ts day (and "d"$a instead of floor a) +/2008.09.09 .k -> .q, 2.4 +/2008.02.03 tick/r.k allow no log +/2007.09.03 check one day flip +/2006.10.18 check type? +/2006.07.24 pub then log +/2006.02.09 fix(2005.11.28) .z.ts end-of-day +/2006.01.05 @[;`sym;`g#] in tick.k load +/2005.12.21 tick/r.k reset `g#sym +/2005.12.11 feed can send .u.endofday +/2005.11.28 zero-end-of-day +/2005.10.28 allow`time on incoming +/2005.10.10 zero latency +/2005.09.30 .u.i->.u.j + +"kdb+tick TICK date:",string[.z.P]," version:",string[.z.K],"_",string .z.k + +/q tick.q SRC [DST] [-p 5010] [-o h] +system"l common/schema.q" +system"l common/log.q" + +if[not system"p";system"p 5010"] + +\l tick/u.q +\d .u +INSTANCE:"one"; +LOGSUFFIX:"tplog"; +ROOT:"data/" +.log.setLogLevel`INFO; + +ld:{if[not type key L::`$(-10_string L),string x;.[L;();:;()]];i::j::-11!(-2;L);if[0<=type i;-2 (string L)," is a corrupt log. Truncate to length ",(string last i)," and restart";exit 1];hopen L}; +tick:{ + init[]; + if[not min(`time`exch`sym~3#key flip value@)each t;'`timeexchym]; + @[;`sym;`g#]each t; + d::.z.D; + if[ l::count x; + L::`$":",ROOT,x,"/log",10#"."; + l::ld d + ] + }; + +endofday:{.log.logInfo"End of day started"; end d;d+:1;if[l;hclose l;l::0(`.u.ld;d)]}; +ts:{if[dtype first x;a,x;(enlist(count first x)#a),x]]; + t insert x;if[l;l enlist (`upd;t;x);j+:1];}]; + +if[not system"t";system"t 1000"; + .log.logInfo"Ticker started in stream mode"; + .z.ts:{ts .z.D}; + upd:{[t;x]ts"d"$a:.z.P; + if[not -12=type first first x;a:"n"$a;x:$[0>type first x;a,x;(enlist(count first x)#a),x]]; + f:key flip value t;pub[t;$[0>type first x;enlist f!x;flip f!x]];if[l;l enlist (`upd;t;x);i+:1];}]; + +\d . +.u.tick[.u.LOGSUFFIX,"_",.u.INSTANCE]; + +.z.ts:{.log.logInfo"Running tick ts";y x}[;oldts:.z.ts] + +\ + globals used + .u.w - dictionary of tables->(handle;syms) + .u.i - msg count in log file + .u.j - total msg count (log file plus those held in buffer) + .u.t - table names + .u.L - tp log filename, e.g. `:./sym2008.09.11 + .u.l - handle to tp log file + .u.d - date + +/test +>q tick.q +>q tick/ssl.q + +/run +>q tick.q sym . -p 5010 /tick +>q tick/r.q :5010 -p 5011 /rdb +>q sym -p 5012 /hdb +>q tick/ssl.q sym :5010 /feed diff --git a/tickplant/tick/r.q b/tickplant/tick/r.q new file mode 100644 index 0000000..a06dcf7 --- /dev/null +++ b/tickplant/tick/r.q @@ -0,0 +1,21 @@ +/q tick/r.q [host]:port[:usr:pwd] [host]:port[:usr:pwd] +/2008.09.09 .k ->.q +system"l common/schema.q" +system"l common/log.q" + +if[not "w"=first string .z.o;system "sleep 1"]; + +upd:insert; + +/ end of day: save, clear, hdb reload +HDBROOT:`$":../db" +.u.end:{t:tables`.;t@:where `g=attr each t@\:`sym;.Q.hdpf[.u.x `hdb;HDBROOT;x;`sym];@[;`sym;`g#] each t;}; + +/ init schema and sync up from log file;cd to hdb(so client save can run) +// log file path comes from .u.L in the ticker +.u.rep:{(.[;();:;].)each x;if[null first y;:()];-11!y;}; + + +// remember to call .u.rep in the client +// this is so clients can choose to override upd before +// calling .u.rep \ No newline at end of file diff --git a/tickplant/tick/u.q b/tickplant/tick/u.q new file mode 100644 index 0000000..a682aa6 --- /dev/null +++ b/tickplant/tick/u.q @@ -0,0 +1,23 @@ +/2019.06.17 ensure sym has g attr for schema returned to new subscriber +/2008.09.09 .k -> .q +/2006.05.08 add +// +\d .u + +init:{w::t!(count t::tables`.)#()} + +del:{w[x]_:w[x;;0]?y};.z.pc:{del[;x]each t}; + +sel:{$[`~y;x;select from x where sym in y]} + +pub:{[t;x]{[t;x;w]if[count x:sel[x]w 1;(neg first w)(`upd;t;x)]}[t;x]each w t} + +add:{$[(count w x)>i:w[x;;0]?.z.w;.[`.u.w;(x;i;1);union;y];w[x],:enlist(.z.w;y)];(x;$[99=type v:value x;sel[v]y;@[0#v;`sym;`g#]])} + +// Sub should return information about the subscription +sub:{if[x~`;:sub[;y]each t];if[not x in t;'x];del[x].z.w;add[x;y]} + +end:{(neg union/[w[;;0]])@\:(`.u.end;x)} + +// simulate some trades happening +ins:{[t;d] if[not t in tables`.;'`unknown_table]; upd[t]select time:.z.p, exch, sym, price, size, side, tradeid:neg[count d]?0Ng from d} diff --git a/todo b/todo new file mode 100644 index 0000000..c4aba0c --- /dev/null +++ b/todo @@ -0,0 +1,3 @@ +optimal attributes +add log lines for connections established/dropped +fix hdb root \ No newline at end of file