trading-go-commons

README

trading-go-commons

Shared Go building blocks for the trading platform's services. These packages consolidate code that was previously copy-pasted across market-data-ingestion, alert-service, context-service, and friends — typed env parsing, a Redis wrapper, a Telegram client, and HTTP server boilerplate.

go get github.com/trogers1052/trading-go-commons

Module path: github.com/trogers1052/trading-go-commons · Go 1.25

Packages

Package	Purpose
env	Typed environment-variable helpers with defaults
redisx	go-redis v9 client with sane defaults + robust retry
telegram	Testable Telegram Bot API client with retry
httpserver	Health/metrics servers + signal-aware context
kafka	sarama-based durable Producer + consumer-group runner

---

env

Pure, typed env-var readers. An unset or empty value always yields the supplied default; values that fail to parse also fall back to the default.

import "github.com/trogers1052/trading-go-commons/env"

cfg := Config{
    Host:        env.String("REDIS_HOST", "localhost"),
    Port:        env.Int("REDIS_PORT", 6379),
    ChatID:      env.Int64("TELEGRAM_CHAT_ID", 0),
    MinConf:     env.Float("MIN_CONFIDENCE", 0.6),
    KafkaOn:     env.Bool("KAFKA_ENABLED", true),
    Brokers:     env.StringSlice("KAFKA_BROKERS", []string{"localhost:19092"}, ","),
    MutedSymbols: env.StringSet("MUTED_SYMBOLS"), // -> map[string]bool, upper-cased
}

StringSlice trims whitespace and drops empty elements; an empty separator defaults to ",". StringSet upper-cases each element and returns nil when the variable is unset.

StringSliceRaw(key, def, sep) is the un-massaged variant: it does a plain strings.Split with no trimming and no dropping of empty elements, mirroring the original strings.Split(getEnv(...), sep) semantics some services relied on. Like StringSlice, an empty separator defaults to ",", and an unset/empty variable returns def unchanged.

// "a, b ,,c" -> ["a", " b ", "", "c"]   (verbatim)
brokers := env.StringSliceRaw("KAFKA_BROKERS", []string{"localhost:19092"}, ",")

---

redisx

A thin wrapper over go-redis v9 that maps Options faithfully onto redis.Options, plus a connection-error classifier and a retry helper.

Zero values pass through. A left-at-zero DialTimeout, ReadTimeout, WriteTimeout, PoolSize, or MinIdleConns means "let go-redis apply its own native default" — this wrapper imposes nothing of its own. Set a field explicitly only when you want to override go-redis's default. (In v0.1.0 the wrapper forced PoolSize: 5 and non-zero timeouts, which differed from go-redis's defaults and had no MinIdleConns field; both are fixed here.)

IsConnectionError is structural — it unwraps the error chain with errors.Is/errors.As against net.Error, io.EOF, and the syscall.ECONNREFUSED/ECONNRESET/EPIPE errnos — instead of fragile substring matching. redis.Nil (key miss) and context cancellation are correctly classified as not connection errors.

import "github.com/trogers1052/trading-go-commons/redisx"

rdb := redisx.NewClient(redisx.Options{
    Addr:     "localhost:6379",
    Password: "",
    DB:       0,
    // Left-zero DialTimeout/ReadTimeout/WriteTimeout/PoolSize/MinIdleConns
    // pass through to go-redis's own native defaults. Set any of them to
    // override, e.g.:
    PoolSize:     20,
    MinIdleConns: 5,
})
defer rdb.Close()

// Retry an operation only when it fails with a connection-level error.
err := redisx.RetryOnConnectionError(ctx, func() error {
    return rdb.Set(ctx, "market.context", payload, 5*time.Minute).Err()
}, 3) // 3 total attempts, exponential backoff (capped at 5s)

RetryOnConnectionError returns immediately on success, returns non-connection errors without retrying (retrying WRONGTYPE can't help), and honours ctx.Done().

---

telegram

The most complete Telegram Bot API client from the platform, generalized. Supports plain and inline-keyboard messages, callback-query acknowledgement, and update polling — all sends retry with exponential backoff. The HTTP client and base URL are injectable, so the client is fully testable with httptest.

import "github.com/trogers1052/trading-go-commons/telegram"

tg := telegram.NewClient(botToken, chatID) // empty token -> sends return an error

// Plain HTML message
_ = tg.SendMessage(ctx, "BUY <b>AAPL</b> — R:R 2.4")

// MarkdownV2 message (caller is responsible for escaping reserved chars)
_ = tg.SendMarkdownMessage(ctx, "*BUY AAPL* — R:R 2\\.4")

// Or pick the parse mode explicitly
_ = tg.SendMessageWithParseMode(ctx, "text", "MarkdownV2")

// Inline keyboard; returns the sent message ID
kb := &telegram.InlineKeyboardMarkup{
    InlineKeyboard: [][]telegram.InlineKeyboardButton{{
        {Text: "Took it",  CallbackData: "took:AAPL"},
        {Text: "Skipped", CallbackData: "skip:AAPL"},
    }},
}
msgID, _ := tg.SendMessageWithKeyboard(ctx, "Trade this setup?", kb)

// Poll for button presses and acknowledge them
updates, _ := tg.GetUpdates(ctx, offset)
for _, u := range updates {
    if cb := u.CallbackQuery; cb != nil {
        _ = tg.AnswerCallbackQuery(ctx, cb.ID, "Logged!")
    }
}

Options for testing / tuning:

tg := telegram.NewClient(token, chatID,
    telegram.WithBaseURL(server.URL),
    telegram.WithHTTPClient(server.Client()),
    telegram.WithMaxRetries(1),
)

---

httpserver

The /health + /metrics + graceful-shutdown boilerplate every service's main.go re-implements, plus a signal-aware context.

import "github.com/trogers1052/trading-go-commons/httpserver"

ctx, stop := httpserver.SignalContext() // cancelled on SIGINT/SIGTERM
defer stop()

health := httpserver.NewHealthServer(":8080")
metrics := httpserver.NewMetricsServer(":9094") // Prometheus default registry
healthErr := health.Start()
metricsErr := metrics.Start()

select {
case <-ctx.Done():
case err := <-healthErr:
    log.Printf("health server: %v", err)
case err := <-metricsErr:
    log.Printf("metrics server: %v", err)
}

_ = health.Shutdown(context.Background())   // 5s default timeout
_ = metrics.Shutdown(context.Background())

Start runs the server in a goroutine and returns a channel that delivers any non-ErrServerClosed error and then closes. Shutdown applies DefaultShutdownTimeout (5s) when the supplied context has no deadline.

The default read/header/write timeouts (5s / 5s / 10s) are conservative; pass functional options to match your prior values:

health := httpserver.NewHealthServer(":8080",
    httpserver.WithReadTimeout(15*time.Second),
    httpserver.WithReadHeaderTimeout(5*time.Second),
    httpserver.WithWriteTimeout(30*time.Second),
    httpserver.WithIdleTimeout(60*time.Second),
)

With no options the behaviour is unchanged from before.

---

kafka

A sarama-based pair of building blocks so every Go service consolidates onto one Kafka library: a durable synchronous Producer and a consumer-group runner (ConsumerGroup) that owns the commit/offset discipline. It replaces the previous sarama/segmentio-kafka-go split while preserving the durability semantics the services already adopted (wait for all in-sync replicas + bounded retry).

IsConnectionError is structural (like redisx's): it unwraps the error chain with errors.Is/errors.As against net.Error, io.EOF, the ECONNREFUSED/ECONNRESET/EPIPE errnos, and sarama's own ErrOutOfBrokers/ErrClosedClient/ErrNotConnected sentinels.

Producer

Durable by default: RequiredAcks = WaitForAll, Producer.Return.Successes = true, Producer.Retry.Max = 3 with backoff, a 10s ack/dial/read/write timeout, and Snappy compression. Publish is synchronous and returns an error only on permanent failure (after sarama's internal retries are exhausted).

import (
    "github.com/IBM/sarama"
    "github.com/trogers1052/trading-go-commons/kafka"
)

prod, err := kafka.NewProducer(
    []string{"localhost:19092"},
    kafka.WithClientID("stock-service"),
    kafka.WithProducerRetries(5),          // Producer.Retry.Max
    kafka.WithProducerRetryBackoff(250*time.Millisecond),
    kafka.WithProducerTimeout(10*time.Second),
    // kafka.WithRequiredAcks(sarama.WaitForLocal), // weaken only on purpose
    // kafka.WithIdempotentProducer(),              // exactly-once per partition
)
if err != nil { /* ... */ }
defer prod.Close()

// key may be nil; non-nil keys partition by key (e.g. symbol).
err = prod.Publish(ctx, "trading.stock.events", []byte("AAPL"), payload)

Producer signatures & options

func NewProducer(brokers []string, opts ...ProducerOption) (*Producer, error)
func (p *Producer) Publish(ctx context.Context, topic string, key, value []byte) error
func (p *Producer) Close() error

WithRequiredAcks(acks sarama.RequiredAcks)   // default WaitForAll
WithProducerRetries(n int)                   // default 3, clamps negatives to 0
WithProducerRetryBackoff(d time.Duration)    // default 250ms
WithProducerTimeout(d time.Duration)         // default 10s (ack + net timeouts)
WithClientID(id string)
WithIdempotentProducer()                     // forces WaitForAll + MaxOpenRequests=1
WithProducerCompression(codec sarama.CompressionCodec) // default Snappy
WithProducerVersion(v sarama.KafkaVersion)

WithIdempotentProducer enables exactly-once-per-partition writes; the constructor automatically pairs it with WaitForAll acks and a single in-flight request, as sarama requires.

ConsumerGroup

NewConsumerGroup joins a group, subscribes to topics, and dispatches each record to a Handler. Run(ctx) blocks until ctx is cancelled, transparently handling session/rebalance cycles (it logs and reconnects after a short backoff on transient session errors) and shutting down gracefully when the context is cancelled.

handler := func(ctx context.Context, msg *kafka.Message) error {
    // msg.Topic, msg.Partition, msg.Offset, msg.Key, msg.Value,
    // msg.Timestamp, msg.Headers (map[string][]byte)
    return process(ctx, msg.Value)
}

cg, err := kafka.NewConsumerGroup(
    []string{"localhost:19092"},
    "alert-service",                  // group ID
    []string{"trading.decisions"},    // topics
    handler,
    kafka.WithInitialOffset(sarama.OffsetOldest), // default Oldest
    kafka.WithOnError(kafka.MarkAndContinue),     // default
)
if err != nil { /* ... */ }
defer cg.Close()

if err := cg.Run(ctx); err != nil { // returns nil on graceful ctx-cancel
    log.Printf("consumer stopped: %v", err)
}

ConsumerGroup signatures & options

type Handler func(ctx context.Context, msg *Message) error

func NewConsumerGroup(brokers []string, groupID string, topics []string,
    handler Handler, opts ...ConsumerOption) (*ConsumerGroup, error)
func (cg *ConsumerGroup) Run(ctx context.Context) error
func (cg *ConsumerGroup) Close() error

WithInitialOffset(offset int64)   // sarama.OffsetOldest (default) | OffsetNewest
WithOnError(policy OnErrorPolicy) // MarkAndContinue (default) | Halt
WithConsumerClientID(id string)
WithConsumerVersion(v sarama.KafkaVersion)

Commit & error policy

The runner owns the commit discipline. After the handler returns for a message:

• Handler success → the message is marked (MarkMessage), advancing the committed offset. Offsets are committed by sarama's auto-commit, so a restarted consumer in the same group does not reprocess marked messages.
• Handler error → behaviour depends on WithOnError:
  • MarkAndContinue (default) — logs the error and still marks the message, then moves on. This prevents a single poison message from wedging the consumer in an infinite reprocess loop; pair it with a dead-letter strategy inside your handler if you need to recover failed payloads.
  • Halt — logs the error, does not mark the message (offset stays put), ends the claim, and Run returns the handler error. The message is redelivered on the next session. Use when no message may ever be skipped.

WithInitialOffset only governs where a brand-new group with no committed offset begins (OffsetOldest reads the backlog; OffsetNewest skips it); once the group has committed offsets, those win.

---

Development

go build ./...
go vet ./...
gofmt -l .                  # must print nothing
go test ./... -short -race -cover   # no Docker: broker tests skip cleanly
go test ./... -race -cover          # with Docker: runs the Redpanda integration tests

The kafka package's integration tests run against a real broker (Redpanda, via trading-testkit's testcontainers helper). Those tests gate on testing.Short(), so -short skips them cleanly when Docker is unavailable; the rest of the suite still runs. The full run (no -short) requires Docker.

CI (GitHub Actions) gates on the gofmt check, go vet, and go test ./... -short -race -cover (no broker required). A separate best-effort integration job runs the full Docker-backed suite but does not block merges. This is a library — there is no Docker/ghcr image build step.

License

MIT — see LICENSE.

---

Built with Claude Code

A large portion of this project — implementation, tests, and documentation — was written in pair-programming sessions with Claude Code, Anthropic's agentic command-line tool.