Merge pull request #52 from Praashh/matching/engine

[feat]: order matching engine

.gitignore

@@ -18,6 +18,9 @@ coverage
 # Turbo
 .turbo
 
+# Snapshot
+snapshot.json
+
 # Vercel
 .vercel
 

docker-compose.yml

@@ -1,8 +1,9 @@
 version: "3.8"
 
 services:
-  db:
-    image: postgres:latest
+  timescaledb:
+    image: timescale/timescaledb:latest-pg12
+    container_name: timescaledb
     ports:
       - 5432:5432
     restart: always
@@ -11,17 +12,18 @@ services:
       POSTGRES_PASSWORD: dev
       POSTGRES_DB: repo
     volumes:
-      - db:/var/lib/postgresql/data
+      - timescale-data:/var/lib/postgresql/data
 
   redis:
-    image: redis:7
+    image: redis:latest
     ports:
       - 6379:6379
     restart: always
     volumes:
       - cache:/data
 
 volumes:
-  db:
+  timescale-data:
   cache:
-    driver: local
+    driver: local
+

packages/order-queue/package.json

@@ -4,6 +4,7 @@
   "main": "index.js",
   "dependencies": {
     "@opinix/logger": "*",
+    "@opinix/types": "*",
     "dotenv": "^16.4.5",
     "nodemon": "^3.1.7",
     "redis": "^4.7.0"

packages/order-queue/src/classes/RedisManager.ts

@@ -0,0 +1,31 @@
+import { RedisClientType, createClient } from "redis";
+import { DbMessage, MessageToApi, WsMessage } from "@opinix/types";
+
+export class RedisManager {
+    private client: RedisClientType;
+    private static instance: RedisManager;
+
+    constructor() {
+        this.client = createClient();
+        this.client.connect();
+    }
+
+    public static getInstance() {
+        if (!this.instance)  {
+            this.instance = new RedisManager();
+        }
+        return this.instance;
+    }
+
+    public pushMessage(message: DbMessage) {
+        this.client.lPush("db_processor", JSON.stringify(message));
+    }
+
+    public publishMessage(channel: string, message: WsMessage) {
+        this.client.publish(channel, JSON.stringify(message));
+    }
+
+    public sendToApi(clientId: string, message: MessageToApi) {
+        this.client.publish(clientId, JSON.stringify(message));
+    }
+}

packages/order-queue/src/index.ts

@@ -1,10 +1,12 @@
 import { addToOrderQueue } from "./queues/orderQueue";
+import { createClient } from "redis";
 import orderWorker from "./queues/orderProcessor";
 import { logger } from "@opinix/logger";
+import { RedisManager } from "./classes/RedisManager";
 const startWorker = async () => {
   logger.info("WORKER | Starting order worker");
   orderWorker;
 };
 
 startWorker();
-export { addToOrderQueue };
+export { addToOrderQueue, RedisManager, createClient };

packages/types/src/index.ts

@@ -1,4 +1,3 @@
-import { OrderStatus } from "@prisma/client";
 export enum EOrderStatus {
   PENDING = "PENDING",
   PLACED = "PLACED",
@@ -18,7 +17,7 @@ export type TOrder = {
   orderBookId: string;
   price: number;
   quantity: number;
-  status: OrderStatus;
+  status: "PLACED" | "PENDING";
   createdAt: Date;
 };
 
@@ -45,3 +44,166 @@ export type TEvent = {
   traders: number;
   quantity: number;
 };
+
+
+
+//  ** Matching Engine Used Types Here ** 
+
+
+export const CREATE_ORDER = "CREATE_ORDER";
+export const CANCEL_ORDER = "CANCEL_ORDER";
+export const TRADE_ADDED = "TRADE_ADDED";
+export const ORDER_UPDATE = "ORDER_UPDATE";
+
+export const ON_RAMP = "ON_RAMP";
+
+export const GET_DEPTH = "GET_DEPTH";
+export const GET_OPEN_ORDERS = "GET_OPEN_ORDERS";
+
+
+//TODO: types sharing between the server, wss and the engine?
+export type MessageFromApi = {
+    type: typeof CREATE_ORDER,
+    data: {
+        market: string,
+        price: number,
+        quantity: number,
+        side: "yes" | "no",
+        userId: string
+    }
+} | {
+    type: typeof CANCEL_ORDER,
+    data: {
+        orderId: string,
+        market: string,
+    }
+} | {
+    type: typeof ON_RAMP,
+    data: {
+        amount: number,
+        userId: string,
+        txnId: string
+    }
+} | {
+    type: typeof GET_DEPTH,
+    data: {
+        market: string,
+    }
+} | {
+    type: typeof GET_OPEN_ORDERS,
+    data: {
+        userId: string,
+        market: string,
+    }
+}
+
+
+
+// *** DB Operation Related Types ***
+
+
+export type DbMessage = {
+  type: typeof TRADE_ADDED,
+  data: {
+      id: string,
+      isBuyerMaker: boolean,
+      price: number,
+      quantity: number,
+      // quoteQuantity: string,
+      timestamp: number,
+      market: string
+  }
+} | {
+  type: typeof ORDER_UPDATE,
+  data: {
+      orderId: string,
+      executedQty: number,
+      market?: string,
+      price?: string,
+      quantity?: string,
+      side?: "yes" | "no",
+  }
+}
+
+
+
+// TYpes for responding the server back
+
+export interface Order {
+  price: number;
+  quantity: number;
+  orderId: string;
+  filled: number;
+  side: "yes" | "no";
+  userId: string;
+}
+
+
+export type MessageToApi = {
+    type: "DEPTH",
+    payload: {
+        bids: [string, string][],
+        asks: [string, string][],
+    }
+} | {
+    type: "ORDER_PLACED",
+    payload: {
+        orderId: string,
+        executedQty: number,
+        fills: {
+            price: number,
+            qty: number,
+            tradeId: string
+        }[]
+    }
+} | {
+    type: "ORDER_CANCELLED",
+    payload: {
+        orderId: string,
+        executedQty: number,
+        remainingQty: number
+    }
+} | {
+    type: "OPEN_ORDERS",
+    payload: Order[]
+}
+
+
+//  WS Types
+
+export type TickerUpdateMessage = {
+  stream: string, 
+  data: {
+      c?: string,
+      h?: string,
+      l?: string,
+      v?: string,
+      V?: string,
+      s?: string,
+      id: number,
+      e: "ticker"
+  }
+}
+
+export type DepthUpdateMessage = {
+  stream: string,
+  data: {
+      b?: [string, string][],
+      a?: [string, string][],
+      e: "depth"
+  }
+}
+
+export type TradeAddedMessage = {
+  stream: string,
+  data: {
+      e: "trade",
+      t: string,
+      m: boolean,
+      p: number,
+      q: string,
+      s: string, // symbol
+  }
+}
+
+export type WsMessage = TickerUpdateMessage | DepthUpdateMessage | TradeAddedMessage;

services/engine/logs/server.log

@@ -0,0 +1,9 @@
+[ 2024-10-20 11:49:07 ] -  info -  WORKER | Starting order worker
+[ 2024-10-20 11:49:07 ] -  info -  SERVER | REDIS: Connected to Redis
+[ 2024-10-20 11:49:07 ] -  info -  SERVER | REDIS: Redis connection is ready to start execution
+[ 2024-10-20 11:51:33 ] -  info -  WORKER | Starting order worker
+[ 2024-10-20 11:51:33 ] -  info -  SERVER | REDIS: Connected to Redis
+[ 2024-10-20 11:51:33 ] -  info -  SERVER | REDIS: Redis connection is ready to start execution
+[ 2024-10-20 12:56:25 ] -  info -  WORKER | Starting order worker
+[ 2024-10-20 12:56:25 ] -  info -  SERVER | REDIS: Connected to Redis
+[ 2024-10-20 12:56:25 ] -  info -  SERVER | REDIS: Redis connection is ready to start execution

services/engine/package.json

@@ -13,18 +13,21 @@
   "description": "",
   "dependencies": {
     "@repo/db": "*",
+    "@opinix/types": "*",
     "@repo/order-queue": "*",
     "@types/cors": "^2.8.17",
     "@types/express": "^4.17.21",
     "body-parser": "^1.20.3",
     "cors": "^2.8.5",
     "dotenv": "^16.4.5",
-    "express": "^4.21.0"
+    "express": "^4.21.0",
+    "uuid": "^10.0.0"
   },
   "devDependencies": {
+    "@types/uuid": "^10.0.0",
     "esbuild": "0.24.0"
   },
-  "exports":{
-    ".":"./src/index.ts"
+  "exports": {
+    ".": "./src/index.ts"
   }
 }

services/engine/src/index.ts

@@ -1 +1,32 @@
-export const ORDERBOOK = {};
+/*
+ TODOS:
+ 1. Fake Liquidity 
+ 2. setting base balancing 
+ 3. Login go through again
+ 4. TEST ENGINE ( Latency ) | shift it to RUST or GO if needed.
+ 5. TYPES
+ 6. Test Cases
+*/
+
+import { createClient } from "@repo/order-queue";
+import { Engine } from "./trade/Engine";
+
+
+async function main() {
+    const engine = new Engine(); 
+    const redisClient = createClient();
+    await redisClient.connect();
+    console.log("connected to redis");
+
+    while (true) {
+        const response = await redisClient.rPop("messages" as string)
+        if (!response) {
+
+        }  else {
+            engine.processOrders(JSON.parse(response));
+        }        
+    }
+
+}
+
+main();