stock-bot/libs/core/event-bus

@stock-bot/event-bus

Lightweight event bus for inter-service communication in the Stock Bot platform.

Overview

This library provides a simple pub/sub event system using Redis, designed for real-time event distribution between microservices. It focuses on simplicity and reliability for event-driven communication.

Features

• Simple pub/sub pattern using Redis
• Automatic reconnection and resubscription
• Local event emission (works even without Redis)
• TypeScript support with predefined trading event types
• Lightweight with minimal dependencies

Installation

bun add @stock-bot/event-bus

Usage

Basic Setup

import { createEventBus, TradingEventType } from '@stock-bot/event-bus';

const eventBus = createEventBus({
  serviceName: 'data-ingestion',
  redisConfig: {
    host: 'localhost',
    port: 6379,
  },
  enableLogging: true,
});

// Wait for connection
await eventBus.waitForConnection();

Publishing Events

// Publish a price update
await eventBus.publish(TradingEventType.PRICE_UPDATE, {
  symbol: 'AAPL',
  price: 150.25,
  volume: 1000000,
  timestamp: Date.now(),
});

// Publish with metadata
await eventBus.publish(TradingEventType.ORDER_FILLED, 
  {
    orderId: '12345',
    symbol: 'TSLA',
    side: 'buy',
    quantity: 100,
    price: 250.50,
  },
  { source: 'ib-gateway', region: 'us' }
);

Subscribing to Events

// Subscribe to price updates
await eventBus.subscribe(TradingEventType.PRICE_UPDATE, async (message) => {
  console.log(`Price update for ${message.data.symbol}: $${message.data.price}`);
});

// Subscribe to order events
await eventBus.subscribe(TradingEventType.ORDER_FILLED, async (message) => {
  const { orderId, symbol, quantity, price } = message.data;
  console.log(`Order ${orderId} filled: ${quantity} ${symbol} @ $${price}`);
});

Event Types

The library includes predefined event types for common trading operations:

enum TradingEventType {
  // Market data events
  PRICE_UPDATE = 'market.price.update',
  ORDERBOOK_UPDATE = 'market.orderbook.update',
  TRADE_EXECUTED = 'market.trade.executed',
  
  // Order events
  ORDER_CREATED = 'order.created',
  ORDER_FILLED = 'order.filled',
  ORDER_CANCELLED = 'order.cancelled',
  ORDER_REJECTED = 'order.rejected',
  
  // Position events
  POSITION_OPENED = 'position.opened',
  POSITION_CLOSED = 'position.closed',
  POSITION_UPDATED = 'position.updated',
  
  // Strategy events
  STRATEGY_SIGNAL = 'strategy.signal',
  STRATEGY_STARTED = 'strategy.started',
  STRATEGY_STOPPED = 'strategy.stopped',
  
  // Risk events
  RISK_LIMIT_BREACH = 'risk.limit.breach',
  RISK_WARNING = 'risk.warning',
  
  // System events
  SERVICE_STARTED = 'system.service.started',
  SERVICE_STOPPED = 'system.service.stopped',
  SERVICE_ERROR = 'system.service.error',
}

Typed Events

Use TypeScript generics for type-safe event handling:

import type { PriceUpdateEvent, OrderEvent } from '@stock-bot/event-bus';

// Type-safe subscription
await eventBus.subscribe<PriceUpdateEvent>(
  TradingEventType.PRICE_UPDATE, 
  async (message) => {
    // message.data is typed as PriceUpdateEvent
    const { symbol, price, volume } = message.data;
  }
);

Cleanup

// Unsubscribe from specific event
await eventBus.unsubscribe(TradingEventType.PRICE_UPDATE);

// Close all connections
await eventBus.close();

Architecture Notes

This library is designed for lightweight, real-time event distribution. For reliable job processing, retries, and persistence, use the @stock-bot/queue library with BullMQ instead.

When to Use Event Bus

• Real-time notifications (price updates, trade executions)
• Service coordination (strategy signals, risk alerts)
• System monitoring (service status, errors)

When to Use Queue

• Data processing jobs
• Batch operations
• Tasks requiring persistence and retries
• Scheduled operations

Error Handling

The event bus handles connection failures gracefully:

try {
  await eventBus.publish(TradingEventType.PRICE_UPDATE, data);
} catch (error) {
  // Event will still be emitted locally
  console.error('Failed to publish to Redis:', error);
}

Development

# Install dependencies
bun install

# Build
bun run build

# Run tests
bun test

# Clean build artifacts
bun run clean