feat(bot-processor): pass candle and candles history to context

packages/backtesting/src/backtesting.ts

@@ -57,7 +57,10 @@ export class Backtesting<T extends IBotConfiguration<T>> {
         // last candle
         await this.processor.stop();
       } else {
-        await this.processor.process();
+        await this.processor.process({
+          candle,
+          candles: candlesticks.slice(0, index + 1),
+        });
       }
 
       const anyOrderPlaced = this.marketSimulator.placeOrders();

packages/bot-processor/src/effect-runner.ts

@@ -12,6 +12,8 @@ import type {
   cancelSmartTrade,
   createSmartTrade,
   replaceSmartTrade,
+  useMarket,
+  useCandle,
 } from "./effects";
 import {
   BUY,
@@ -24,6 +26,8 @@ import {
   USE_INDICATORS,
   USE_SMART_TRADE,
   USE_TRADE,
+  USE_MARKET,
+  USE_CANDLE,
 } from "./effects";
 
 export const effectRunnerMap: Record<
@@ -42,6 +46,8 @@ export const effectRunnerMap: Record<
   [USE_INDICATORS]: () => {
     throw new Error("useIndicators() hook is deprecated");
   },
+  [USE_MARKET]: runUseMarketEffect,
+  [USE_CANDLE]: runUseCandleEffect,
 };
 
 async function runUseSmartTradeEffect(
@@ -193,3 +199,23 @@ async function runUseExchangeEffect(
 
   return ctx.exchange;
 }
+
+async function runUseMarketEffect(
+  _effect: ReturnType<typeof useMarket>,
+  ctx: TBotContext<any>,
+) {
+  return ctx.market;
+}
+
+async function runUseCandleEffect(
+  effect: ReturnType<typeof useCandle>,
+  ctx: TBotContext<any>,
+) {
+  const index = effect.payload;
+
+  if (index >= 0) {
+    return ctx.market.candles[index];
+  }
+
+  return ctx.market.candles[ctx.market.candles.length + index];
+}

packages/bot-processor/src/effects/index.ts

@@ -6,3 +6,4 @@ export * from "./sell";
 export * from "./useExchange";
 export * from "./useIndicators";
 export * from "./useTrade";
+export * from "./market";

packages/bot-processor/src/effects/market.ts

@@ -0,0 +1,15 @@
+import { USE_MARKET, USE_CANDLE } from "./types";
+import { makeEffect } from "./utils";
+
+export function useMarket() {
+  return makeEffect(USE_MARKET, undefined, undefined);
+}
+
+/**
+ * Get candle data for the given index.
+ * If the index is negative, it will return the candle data from the end.
+ * By default, will return the last candle.
+ */
+export function useCandle(index = -1) {
+  return makeEffect(USE_CANDLE, index, undefined);
+}

packages/bot-processor/src/effects/types/base-effect.ts

@@ -1,4 +1,4 @@
-import { EffectType } from "./effect-types";
+import type { EffectType } from "./effect-types";
 
 export type BaseEffect<T extends EffectType, P = undefined, R = undefined> = {
   type: T;

packages/bot-processor/src/effects/types/effect-types.ts

@@ -8,6 +8,8 @@ export const CREATE_SMART_TRADE = "CREATE_SMART_TRADE";
 export const CANCEL_SMART_TRADE = "CANCEL_SMART_TRADE";
 export const USE_EXCHANGE = "USE_EXCHANGE";
 export const USE_INDICATORS = "USE_INDICATORS";
+export const USE_MARKET = "USE_MARKET";
+export const USE_CANDLE = "USE_CANDLE";
 
 export type EffectType =
   | typeof USE_SMART_TRADE
@@ -19,4 +21,6 @@ export type EffectType =
   | typeof CREATE_SMART_TRADE
   | typeof CANCEL_SMART_TRADE
   | typeof USE_EXCHANGE
-  | typeof USE_INDICATORS;
+  | typeof USE_INDICATORS
+  | typeof USE_MARKET
+  | typeof USE_CANDLE;

packages/bot-processor/src/strategy-runner.ts

@@ -7,6 +7,7 @@ import type {
   IBotConfiguration,
   IBotControl,
   IStore,
+  MarketData,
   TBotContext,
 } from "./types";
 import { createContext } from "./utils/createContext";
@@ -41,12 +42,13 @@ export class StrategyRunner<T extends IBotConfiguration> {
     await this.runTemplate(context);
   }
 
-  async process() {
+  async process(market?: MarketData) {
     const context = createContext(
       this.control,
       this.botConfig,
       this.exchange,
       "process",
+      market,
     );
 
     await this.runTemplate(context);

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

@@ -1,4 +1,5 @@
 import type { IExchange } from "@opentrader/exchanges";
+import type { MarketData } from "../market";
 import type { IBotControl } from "./bot-control.interface";
 import type { IBotConfiguration } from "./bot-configuration.interface";
 
@@ -22,4 +23,8 @@ export type TBotContext<T extends IBotConfiguration> = {
   onStart: boolean;
   onStop: boolean;
   onProcess: boolean;
+  /**
+   * Marked data
+   */
+  market: MarketData;
 };

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

@@ -1,3 +1,4 @@
 export * from "./bot";
 export * from "./smart-trade";
 export * from "./store";
+export * from "./market";

packages/bot-processor/src/types/market/index.ts

@@ -0,0 +1 @@
+export * from "./market-data";

packages/bot-processor/src/types/market/market-data.ts

@@ -0,0 +1,12 @@
+import type { ICandlestick } from "@opentrader/types";
+
+export interface MarketData {
+  /**
+   * Lst closed candlestick
+   */
+  candle?: ICandlestick;
+  /**
+   * List of previous candles, included last one. Last candle can be accessed by `candles[candles.length - 1]`
+   */
+  candles: ICandlestick[];
+}

packages/bot-processor/src/utils/createContext.ts

@@ -1,11 +1,19 @@
 import type { IExchange } from "@opentrader/exchanges";
-import type { IBotConfiguration, IBotControl, TBotContext } from "../types";
+import type {
+  IBotConfiguration,
+  IBotControl,
+  MarketData,
+  TBotContext,
+} from "../types";
 
 export function createContext<T extends IBotConfiguration>(
   control: IBotControl,
   config: T,
   exchange: IExchange,
   command: "start" | "stop" | "process", // @todo add type in file
+  market: MarketData = {
+    candles: [],
+  },
 ): TBotContext<T> {
   return {
     control,
@@ -15,5 +23,6 @@ export function createContext<T extends IBotConfiguration>(
     onStart: command === "start",
     onStop: command === "stop",
     onProcess: command === "process",
+    market,
   };
 }

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

@@ -0,0 +1,28 @@
+import type { IExchange } from "@opentrader/exchanges";
+import type { MarketData, TBotContext } from "@opentrader/bot-processor";
+import { useMarket, useCandle, useExchange } from "@opentrader/bot-processor";
+import { logger } from "@opentrader/logger";
+
+export function* testCandle(ctx: TBotContext<any>) {
+  const { config: bot, onStart, onStop } = ctx;
+  const exchange: IExchange = yield useExchange();
+
+  if (onStart) {
+    logger.info(
+      `[CandleStrategy] Bot started. Using ${exchange.exchangeCode} exchange`,
+    );
+    return;
+  }
+  if (onStop) {
+    logger.info(`[CandleStrategy] Bot stopped`);
+    return;
+  }
+
+  const market: MarketData = yield useMarket();
+  logger.info(market, `[CandleStrategy] Market data`);
+
+  const candle: MarketData["candle"] = yield useCandle();
+  logger.info(`[CandleStrategy] Candle ${JSON.stringify(candle)}`);
+
+  logger.info(`[CandleStrategy] Bot template executed successfully`);
+}

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

@@ -1,3 +1,4 @@
 export * from "./buySell";
 export * from "./trade";
 export * from "./debug";
+export * from "./candle";

packages/bot/src/channels/candles/candles.aggregator.ts

@@ -8,7 +8,7 @@ import type { CandlesWatcher } from "./candles.watcher";
  * Aggregates 1m candles to higher timeframes.
  *
  * Emits:
- * - candle: `ICandlestick`
+ * - candle(lastCandle: `ICandlestick`, candlesHistory: ICandlestick[]): void
  */
 export class CandlesAggregator extends EventEmitter {
   public timeframe: BarSize;
@@ -21,6 +21,10 @@ export class CandlesAggregator extends EventEmitter {
    * Storing 1m candles for further aggregation.
    */
   private bucket: ICandlestick[] = [];
+  /**
+   * Pushing aggregated candles to history.
+   */
+  private candlesHistory: ICandlestick[] = [];
   private candlesWatcher: CandlesWatcher;
 
   constructor(timeframe: BarSize, candlesWatcher: CandlesWatcher) {
@@ -39,12 +43,13 @@ export class CandlesAggregator extends EventEmitter {
           `Bucket length of ${this.symbol} reached ${this.bucket.length}/${this.bucketSize}. Aggregating ${this.timeframe} bucket`,
         );
         const candle = this.aggregate();
+        this.candlesHistory.push(candle);
 
         logger.info(
           candle,
           `Aggregated ${this.symbol} ${this.bucketSize}m candles to ${this.timeframe}: O: ${candle.open}, H: ${candle.high}, L: ${candle.low}, C: ${candle.close} at ${new Date(candle.timestamp).toISOString()}`,
         );
-        this.emit("candle", candle);
+        this.emit("candle", candle, this.candlesHistory);
 
         return;
       }

packages/bot/src/channels/candles/candles.channel.ts

@@ -63,11 +63,12 @@ export class CandlesChannel extends EventEmitter {
     }
 
     aggregator = new CandlesAggregator(timeframe, watcher);
-    aggregator.on("candle", (candle: ICandlestick) => {
+    aggregator.on("candle", (candle: ICandlestick, history: ICandlestick[]) => {
       const candleEvent: CandleEvent = {
         symbol,
         timeframe,
         candle,
+        history,
       };
 
       this.emit("candle", candleEvent);

packages/bot/src/channels/candles/types.ts

@@ -3,5 +3,12 @@ import type { BarSize, ICandlestick } from "@opentrader/types";
 export type CandleEvent = {
   symbol: string;
   timeframe: BarSize;
+  /**
+   * Last closed candle
+   */
   candle: ICandlestick;
+  /**
+   * Candles history
+   */
+  history: ICandlestick[];
 };

packages/bot/src/processing/candles.processor.ts

@@ -62,7 +62,7 @@ export class CandlesProcessor {
 
   // @todo maybe queue
   private async handleCandle(data: CandleEvent) {
-    const { candle, symbol, timeframe } = data;
+    const { candle, history, symbol, timeframe } = data;
 
     logger.info(
       `CandlesProcessor: Received candle ${timeframe} for ${symbol}. Start processing.`,
@@ -85,7 +85,10 @@ export class CandlesProcessor {
         continue;
       }
 
-      await botProcessor.process();
+      await botProcessor.process({
+        candle,
+        candles: history,
+      });
       await botProcessor.placePendingOrders();
 
       logger.info(`Exec bot #${bot.id} template done`);

packages/processing/src/bot/bot.processing.ts

@@ -1,4 +1,4 @@
-import type { IBotConfiguration } from "@opentrader/bot-processor";
+import type { IBotConfiguration, MarketData } from "@opentrader/bot-processor";
 import { createStrategyRunner } from "@opentrader/bot-processor";
 import { findTemplate } from "@opentrader/bot-templates";
 import { exchangeProvider } from "@opentrader/exchanges";
@@ -69,8 +69,13 @@ export class BotProcessing {
     });
   }
 
-  async processCommand(command: "start" | "stop" | "process") {
-    console.log(`🤖 Bot #${this.bot.id} command=${command}`);
+  async processCommand(
+    command: "start" | "stop" | "process",
+    market?: MarketData,
+  ) {
+    console.log(
+      `🤖 Bot #${this.bot.id} command=${command} candle=${JSON.stringify(market?.candle)} candlesHistory=${market?.candles.length} start`,
+    );
     if (this.isBotProcessing()) {
       console.warn(
         `Cannot execute "${command}()" command. The bot is busy right now by the previous processing job.`,
@@ -87,7 +92,7 @@ export class BotProcessing {
       } else if (command === "stop") {
         await processor.stop();
       } else if (command === "process") {
-        await processor.process();
+        await processor.process(market);
       }
     } catch (err) {
       await xprisma.bot.setProcessing(false, this.bot.id);
@@ -107,8 +112,8 @@ export class BotProcessing {
     await this.processCommand("stop");
   }
 
-  async process() {
-    await this.processCommand("process");
+  async process(market?: MarketData) {
+    await this.processCommand("process", market);
   }
 
   isBotRunning() {