redis

Documentation

Overview

Package redis provides Redis Streams integration for the liquidation heatmap service

Index

• func ParseLiquidationMessage(msg redis.XMessage) (*models.LiquidationEvent, error)
• type Config
• type ConsumerConfig
• type StreamClient
  • func NewStreamClient(cfg Config) (*StreamClient, error)
  • func (s *StreamClient) CacheAndPublishHeatmap(heatmap *models.HeatmapData) error
  • func (s *StreamClient) Close() error
  • func (s *StreamClient) ConsumeHeatmapStream(symbol models.Symbol, handler func(*models.HeatmapData) error) error
  • func (s *StreamClient) ConsumeLiquidationStream(cfg ConsumerConfig, handler func(*models.LiquidationEvent) error) error
  • func (s *StreamClient) GetCachedHeatmap(symbol models.Symbol, interval models.Interval) (*models.HeatmapData, error)
  • func (s *StreamClient) GetLatestHeatmaps(symbols []models.Symbol, interval models.Interval) (map[models.Symbol]*models.HeatmapData, error)
  • func (s *StreamClient) GetPendingMessages(streamName, group string) (*redis.XPending, error)
  • func (s *StreamClient) GetStats() (map[string]interface{}, error)
  • func (s *StreamClient) GetStreamInfo(streamName string) (*redis.XInfoStream, error)
  • func (s *StreamClient) GetStreamLength(streamName string) (int64, error)
  • func (s *StreamClient) HealthCheck() error
  • func (s *StreamClient) Ping() (bool, error)
  • func (s *StreamClient) PublishLiquidation(event *models.LiquidationEvent) error
  • func (s *StreamClient) PublishLiquidationBatch(events []*models.LiquidationEvent) error
  • func (s *StreamClient) PublishMarketSnapshot(snapshot *models.MarketSnapshot) error
  • func (s *StreamClient) PublishOrderBook(ob *models.OrderBookSnapshot) error
  • func (s *StreamClient) TrimStream(streamName string, maxLen int64) error

Functions

func ParseLiquidationMessage

func ParseLiquidationMessage(msg redis.XMessage) (*models.LiquidationEvent, error)

ParseLiquidationMessage parses a liquidation message from Redis

Types

type Config

type Config struct {
	Host            string
	Port            int
	Password        string
	DB              int
	StreamMaxLen    int64         // Max messages per stream (default: 10000)
	HeatmapCacheTTL time.Duration // TTL for cached heatmaps (default: 5 minutes)
}

Config holds Redis configuration

type ConsumerConfig

type ConsumerConfig struct {
	Group     string
	Consumer  string
	Streams   []string
	BatchSize int64
	BlockMs   int64
}

ConsumerConfig holds consumer configuration

type StreamClient

type StreamClient struct {
	// contains filtered or unexported fields
}

StreamClient handles Redis Streams operations

func NewStreamClient

func NewStreamClient(cfg Config) (*StreamClient, error)

NewStreamClient creates a new Redis Streams client

func (*StreamClient) CacheAndPublishHeatmap

func (s *StreamClient) CacheAndPublishHeatmap(heatmap *models.HeatmapData) error

CacheAndPublishHeatmap stores heatmap in cache AND publishes to stream

func (*StreamClient) Close

func (s *StreamClient) Close() error

Close closes the Redis connection

func (*StreamClient) ConsumeHeatmapStream

func (s *StreamClient) ConsumeHeatmapStream(symbol models.Symbol, handler func(*models.HeatmapData) error) error

ConsumeHeatmapStream consumes heatmap updates from stream

func (*StreamClient) ConsumeLiquidationStream

func (s *StreamClient) ConsumeLiquidationStream(cfg ConsumerConfig, handler func(*models.LiquidationEvent) error) error

ConsumeLiquidationStream consumes liquidation events from stream

func (*StreamClient) GetCachedHeatmap

func (s *StreamClient) GetCachedHeatmap(symbol models.Symbol, interval models.Interval) (*models.HeatmapData, error)

GetCachedHeatmap retrieves cached heatmap data

func (*StreamClient) GetLatestHeatmaps

func (s *StreamClient) GetLatestHeatmaps(symbols []models.Symbol, interval models.Interval) (map[models.Symbol]*models.HeatmapData, error)

GetLatestHeatmaps retrieves the most recent heatmap for each symbol

func (*StreamClient) GetPendingMessages

func (s *StreamClient) GetPendingMessages(streamName, group string) (*redis.XPending, error)

GetPendingMessages returns pending messages for a consumer group

func (*StreamClient) GetStats

func (s *StreamClient) GetStats() (map[string]interface{}, error)

GetStats returns statistics about all relevant streams

func (*StreamClient) GetStreamInfo

func (s *StreamClient) GetStreamInfo(streamName string) (*redis.XInfoStream, error)

GetStreamInfo returns information about a stream

func (*StreamClient) GetStreamLength

func (s *StreamClient) GetStreamLength(streamName string) (int64, error)

GetStreamLength returns the number of messages in a stream

func (*StreamClient) HealthCheck

func (s *StreamClient) HealthCheck() error

HealthCheck performs a health check on Redis connection

func (*StreamClient) Ping

func (s *StreamClient) Ping() (bool, error)

Ping checks Redis connection status

func (*StreamClient) PublishLiquidation

func (s *StreamClient) PublishLiquidation(event *models.LiquidationEvent) error

PublishLiquidation publishes a liquidation event to the stream

func (*StreamClient) PublishLiquidationBatch

func (s *StreamClient) PublishLiquidationBatch(events []*models.LiquidationEvent) error

PublishLiquidationBatch publishes multiple liquidation events efficiently

func (*StreamClient) PublishMarketSnapshot

func (s *StreamClient) PublishMarketSnapshot(snapshot *models.MarketSnapshot) error

PublishMarketSnapshot publishes market data to the stream

func (*StreamClient) PublishOrderBook

func (s *StreamClient) PublishOrderBook(ob *models.OrderBookSnapshot) error

PublishOrderBook publishes order book snapshot to the stream

func (*StreamClient) TrimStream

func (s *StreamClient) TrimStream(streamName string, maxLen int64) error

TrimStream trims a stream to a maximum length