feat(Trade): implement Stop Loss (#89)

packages/bot-processor/src/effects/smart-trade.ts

@@ -23,6 +23,11 @@ export type UseSmartTradePayload = {
     status?: OrderStatusEnum; // default to Idle
     price?: number; // if undefined, then it's a market order
   };
+  sl?: {
+    type: OrderType;
+    price?: number; // if undefined, then it's a market order
+    stopPrice: number;
+  };
   quantity: number;
 };
 
@@ -34,10 +39,7 @@ export function getSmartTrade(ref = DEFAULT_REF) {
   return makeEffect(GET_SMART_TRADE, undefined, ref);
 }
 
-export function createSmartTrade(
-  payload: UseSmartTradePayload,
-  ref = DEFAULT_REF,
-) {
+export function createSmartTrade(payload: UseSmartTradePayload, ref = DEFAULT_REF) {
   return makeEffect(CREATE_SMART_TRADE, payload, ref);
 }
 

packages/bot-processor/src/types/bot/bot-template.type.ts

@@ -93,5 +93,6 @@ export interface BotTemplate<T extends IBotConfiguration> {
     [MarketEventType.onOrderbookChange]?: boolean | ((botConfig: T) => boolean);
     [MarketEventType.onTickerChange]?: boolean | ((botConfig: T) => boolean);
     [MarketEventType.onOrderFilled]?: boolean | ((botConfig: T) => boolean);
+    [MarketEventType.onTradeCompleted]?: boolean | ((botConfig: T) => boolean);
   };
 }

packages/bot-processor/src/types/smart-trade/smart-trade.type.ts

@@ -4,6 +4,7 @@ type OrderBuilder<T extends OrderType, S extends OrderStatusEnum> = {
   type: T;
   status: S;
   price: T extends "Limit" ? number : undefined;
+  stopPrice?: number;
   filledPrice: S extends "filled" ? number : undefined;
   /**
    * Creation time, Unix timestamp format in milliseconds, e.g. `1597026383085`
@@ -33,6 +34,7 @@ type SmartTradeBuilder<WithSell extends boolean> = {
   quantity: number;
   buy: Order;
   sell: WithSell extends true ? Order : undefined;
+  sl?: Order;
   type: XSmartTradeType;
 };
 

packages/bot-processor/src/types/store/types.ts

@@ -11,6 +11,7 @@ export type OrderPayload = {
   symbol?: string;
   type: OrderType;
   status?: OrderStatusEnum; // default to Idle
+  stopPrice?: number;
   price?: number; // if undefined, then it's a market order
   /**
    * Price deviation relative to entry price.
@@ -30,6 +31,7 @@ export type CreateSmartTradePayload = {
   type: XSmartTradeType;
   buy: OrderPayload;
   sell?: OrderPayload;
+  sl?: OrderPayload;
   additionalOrders?: AdditionalOrderPayload[];
   quantity: number;
 };

packages/bot-templates/src/templates/grid-bot.ts

@@ -71,7 +71,7 @@ gridBot.schema = z.object({
   ),
 });
 gridBot.runPolicy = {
-  onOrderFilled: true,
+  onTradeCompleted: true,
 };
 
 export type GridBotConfig = IBotConfiguration<z.infer<typeof gridBot.schema>>;

packages/bot-templates/src/templates/grid.ts

@@ -41,7 +41,7 @@ grid.schema = z.object({
   quantityPerGrid: z.number().positive().describe("Quantity of base currency per each grid"),
 });
 grid.runPolicy = {
-  onOrderFilled: true,
+  onTradeCompleted: true,
 };
 
 export type GridBotLiteConfig = IBotConfiguration<z.infer<typeof grid.schema>>;

packages/bot-templates/src/templates/test/index.ts

@@ -7,3 +7,4 @@ export * from "./state.js";
 export * from "./trades.js";
 export * from "./testMarketOrder.js";
 export * from "./dca.js";
+export * from "./stopLoss.js";

packages/bot-templates/src/templates/test/stopLoss.ts

@@ -0,0 +1,64 @@
+import { z } from "zod";
+import {
+  cancelSmartTrade,
+  IBotConfiguration,
+  type SmartTradeService,
+  type TBotContext,
+  useSmartTrade,
+} from "@opentrader/bot-processor";
+import { logger } from "@opentrader/logger";
+
+export function* testStopLoss(ctx: TBotContext<TestStopLossSchema>) {
+  logger.info("[TestStopLoss] Executing strategy template");
+  const { entry, tp, sl } = ctx.config.settings;
+
+  if (ctx.onStop) {
+    logger.info("[TestStopLoss] Stopping strategy");
+    yield cancelSmartTrade();
+    return;
+  }
+
+  if (ctx.onStart) {
+    const smartTrade: SmartTradeService = yield useSmartTrade({
+      buy: {
+        type: entry.type,
+      },
+      sell: {
+        type: tp.type,
+        price: tp.price,
+      },
+      sl: {
+        type: sl.type,
+        stopPrice: sl.stopPrice,
+        price: sl.price,
+      },
+      quantity: 0.0001,
+    });
+    logger.info(smartTrade, "[TestStopLoss] Trade created");
+  }
+}
+
+testStopLoss.schema = z.object({
+  entry: z.object({
+    type: z.enum(["Market", "Limit"]).default("Market").describe("Type of the entry order."),
+    price: z.number().optional().describe("Limit price of the entry order."),
+  }),
+  tp: z.object({
+    type: z.enum(["Limit", "Market"]).default("Limit").describe("Type of the take profit order."),
+    price: z.number().default(105000).describe("Limit price of the take profit order."),
+  }),
+  sl: z.object({
+    type: z.enum(["Limit", "Market"]).default("Limit").describe("Type of the stop loss order."),
+    stopPrice: z.number().default(95000).describe("Stop price of the stop loss order."),
+    price: z.number().default(90000).describe("Limit price of the stop loss order."),
+  }),
+});
+testStopLoss.hidden = true;
+testStopLoss.watchers = {
+  watchCandles: ({ symbol }: IBotConfiguration) => symbol,
+};
+testStopLoss.runPolicy = {
+  onCandleClosed: true,
+};
+
+export type TestStopLossSchema = IBotConfiguration<z.infer<typeof testStopLoss.schema>>;

packages/bot/package.json

@@ -30,6 +30,7 @@
     "@opentrader/tools": "workspace:*",
     "async": "^3.2.6",
     "ccxt": "4.4.41",
-    "cron": "^3.3.1"
+    "cron": "^3.3.1",
+    "emittery": "^1.0.3"
   }
 }

packages/bot/src/bot.manager.ts

@@ -17,7 +17,7 @@ export class Bot {
   ) {}
 
   async start() {
-    eventBus.beforeBotStarted(this.bot);
+    await eventBus.emit("onBeforeBotStarted", this.bot);
 
     // 1. Exec "start" on the strategy fn
     const botProcessor = new BotProcessing(this.bot);
@@ -29,28 +29,35 @@ export class Bot {
       include: { exchangeAccount: true },
     });
 
-    // 2. Subscribe to Market and Order events
+    // 2. Place pending trades
+    const pendingSmartTrades = await botProcessor.getPendingSmartTrades();
+    for (const trade of pendingSmartTrades) {
+      await eventBus.emit("onTradeCreated", trade);
+    }
+
+    // 3. Subscribe to Market and Order events
     await this.watchStreams();
 
-    eventBus.botStarted(this.bot);
+    await eventBus.emit("onBotStarted", this.bot);
   }
 
   async stop() {
+    await eventBus.emit("onBeforeBotStopped", this.bot);
+
     // 1. Unsubscribe from Market and Order events
     this.unwatchStreams();
 
-    eventBus.beforeBotStopped(this.bot);
-
     // 2. Exec "stop" on the strategy fn
     const botProcessor = new BotProcessing(this.bot);
     await botProcessor.processStopCommand();
+
     this.bot = await xprisma.bot.custom.update({
       where: { id: this.bot.id },
       data: { enabled: false },
       include: { exchangeAccount: true },
     });
 
-    eventBus.botStopped(this.bot);
+    await eventBus.emit("onBotStopped", this.bot);
   }
 
   async watchStreams() {

packages/bot/src/platform.ts

@@ -1,24 +1,35 @@
 import { findStrategy, loadCustomStrategies } from "@opentrader/bot-templates/server";
-import { xprisma, type ExchangeAccountWithCredentials, TBotWithExchangeAccount } from "@opentrader/db";
+import {
+  xprisma,
+  type ExchangeAccountWithCredentials,
+  TBotWithExchangeAccount,
+  SmartTradeWithOrders,
+} from "@opentrader/db";
 import { logger } from "@opentrader/logger";
 import { exchangeProvider } from "@opentrader/exchanges";
 import { BotProcessing } from "@opentrader/processing";
 import { eventBus } from "@opentrader/event-bus";
 import { store } from "@opentrader/bot-store";
 import { MarketEvent } from "@opentrader/types";
+import { EventEmitter } from "node:events";
 import { Bot } from "./bot.manager.js";
 
 import { MarketsStream } from "./streams/markets.stream.js";
-import { OrdersStream } from "./streams/orders.stream.js";
+import { OrderEvent, OrdersStream } from "./streams/orders.stream.js";
+import { TradeManager } from "./trade.manager.js";
 
 export class Platform {
   private ordersStream: OrdersStream;
   private marketStream: MarketsStream;
   private unsubscribeFromEventBus = () => {};
   private enabledBots: Bot[] = [];
+  private trades: TradeManager[] = [];
 
   constructor(exchangeAccounts: ExchangeAccountWithCredentials[]) {
+    EventEmitter.defaultMaxListeners = 0; // Disable Node.js max listeners warning
+
     this.ordersStream = new OrdersStream(exchangeAccounts);
+    this.ordersStream.on("order", this.handleOrderEvent);
 
     this.marketStream = new MarketsStream(this.enabledBots.map(({ bot }) => bot));
     this.marketStream.on("market", this.handleMarketEvent);
@@ -136,21 +147,36 @@ export class Platform {
     const onBeforeBotStarted = async (_data: TBotWithExchangeAccount) => {
       //
     };
-    const onBotStarted = async (data: TBotWithExchangeAccount) => {
+
+    const startBot = async (data: TBotWithExchangeAccount) => {
       const bot = new Bot(data, this.marketStream, this.ordersStream);
-      await bot.watchStreams();
+      await bot.start();
       this.enabledBots.push(bot);
     };
 
-    const onBeforeBotStopped = async (data: TBotWithExchangeAccount) => {
+    const onBotStarted = async (_data: TBotWithExchangeAccount) => {
+      //
+    };
+
+    const stopBot = async (data: TBotWithExchangeAccount) => {
       const bot = this.enabledBots.find(({ bot }) => bot.id === data.id);
+
       if (bot) {
-        bot.unwatchStreams();
-      } else {
-        logger.warn(`onBeforeBotStopped: Bot not found [id=${data.id} name=${data.name}]`);
+        const botTrades = this.trades.filter((trade) => trade.smartTrade.botId === bot.bot.id);
+        for (const trade of botTrades) {
+          trade.unwatchStreams();
+        }
+        this.trades = this.trades.filter(({ smartTrade }) => smartTrade.botId !== bot.bot.id);
+
+        await bot.stop();
+        this.enabledBots = this.enabledBots.filter((enabledBot) => enabledBot !== bot);
       }
     };
 
+    const onBeforeBotStopped = async (_data: TBotWithExchangeAccount) => {
+      //
+    };
+
     const onBotStopped = async (data: TBotWithExchangeAccount) => {
       this.enabledBots = this.enabledBots.filter(({ bot }) => bot.id !== data.id);
       await this.marketStream.clean(this.enabledBots.map(({ bot }) => bot));
@@ -169,25 +195,53 @@ export class Platform {
       await this.ordersStream.updateExchangeAccount(exchangeAccount);
     };
 
+    const onTradeCreated = async (trade: SmartTradeWithOrders) => {
+      let tradeManager = this.trades.find(({ smartTrade }) => smartTrade.id === trade.id);
+      if (!tradeManager) {
+        tradeManager = new TradeManager(trade, this.ordersStream);
+        this.trades.push(tradeManager);
+      }
+
+      await tradeManager.next();
+    };
+
+    const onTradeCompleted = async (trade: SmartTradeWithOrders) => {
+      this.trades = this.trades.filter(({ smartTrade }) => smartTrade.id !== trade.id);
+    };
+
+    eventBus.on("startBot", startBot);
     eventBus.on("onBeforeBotStarted", onBeforeBotStarted);
     eventBus.on("onBotStarted", onBotStarted);
+    eventBus.on("stopBot", stopBot);
     eventBus.on("onBeforeBotStopped", onBeforeBotStopped);
     eventBus.on("onBotStopped", onBotStopped);
     eventBus.on("onExchangeAccountCreated", addExchangeAccount);
     eventBus.on("onExchangeAccountDeleted", removeExchangeAccount);
     eventBus.on("onExchangeAccountUpdated", updateExchangeAccount);
+    eventBus.on("onTradeCreated", onTradeCreated);
+    eventBus.on("onTradeCompleted", onTradeCompleted);
 
     // Return unsubscribe function
     return () => {
+      eventBus.off("startBot", startBot);
+      eventBus.off("onBeforeBotStarted", onBeforeBotStarted);
       eventBus.off("onBotStarted", onBotStarted);
+      eventBus.off("stopBot", stopBot);
+      eventBus.off("onBeforeBotStopped", onBeforeBotStopped);
       eventBus.off("onBotStopped", onBotStopped);
       eventBus.off("onExchangeAccountCreated", addExchangeAccount);
       eventBus.off("onExchangeAccountDeleted", removeExchangeAccount);
       eventBus.off("onExchangeAccountUpdated", updateExchangeAccount);
+      eventBus.off("onTradeCreated", onTradeCreated);
+      eventBus.off("onTradeCompleted", onTradeCompleted);
     };
   }
 
   handleMarketEvent = (event: MarketEvent) => {
     store.updateMarket(event);
   };
+
+  handleOrderEvent = async (_event: OrderEvent) => {
+    //
+  };
 }

packages/bot/src/queue/queue.ts

@@ -1,3 +1,4 @@
+import { eventBus } from "@opentrader/event-bus";
 import { cargoQueue, QueueObject } from "async";
 import type { TBot } from "@opentrader/db";
 import { BotProcessing } from "@opentrader/processing";
@@ -22,7 +23,10 @@ async function queueHandler(tasks: QueueEvent[]) {
     markets: store.getMarkets(event.subscribedMarkets),
   });
 
-  await botProcessor.placePendingOrders();
+  const pendingSmartTrades = await botProcessor.getPendingSmartTrades();
+  for (const trade of pendingSmartTrades) {
+    await eventBus.emit("onTradeCreated", trade);
+  }
 }
 
 const createQueue = () => cargoQueue<QueueEvent>(queueHandler);

packages/bot/src/queue/types.ts

@@ -1,4 +1,4 @@
-import type { TBotWithExchangeAccount } from '@opentrader/db';
+import type { TBotWithExchangeAccount } from "@opentrader/db";
 import { MarketEvent, MarketId, MarketEventType } from "@opentrader/types";
 
 export type OrderFilledEvent = {
@@ -7,6 +7,12 @@ export type OrderFilledEvent = {
   orderId: number;
 };
 
-export type ProcessingEvent = MarketEvent | OrderFilledEvent;
+export type TradeCompletedEvent = {
+  type: "onTradeCompleted";
+  marketId: MarketId;
+  tradeId: number;
+};
+
+export type ProcessingEvent = MarketEvent | OrderFilledEvent | TradeCompletedEvent;
 
 export type QueueEvent = ProcessingEvent & { bot: TBotWithExchangeAccount; subscribedMarkets: MarketId[] };