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	kafka-go-based durable Producer (headers + batch) + consumer-group runner
clock	The time source a service reads "now" from — real by default, simulated under replay

---

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 segmentio/kafka-go-based pair of building blocks so every Go service consolidates onto one Kafka library (pure Go, lighter GC than sarama, simple multi-arch Pi builds): a durable synchronous Producer (with record headers and batch publish) and a consumer-group runner (ConsumerGroup) that owns the manual commit/offset discipline.

The public API is library-neutral — callers pass package-defined enums (RequiredAcks, OffsetPosition, OnErrorPolicy, Compression), never kafka-go (or sarama) types. kafka-go types do not leak through the public surface.

Breaking change (v0.3.0 → v0.4.0). The package moved from sarama to kafka-go. WithInitialOffset now takes kafka.OffsetOldest / kafka.OffsetNewest (was sarama.OffsetOldest/Newest), and WithRequiredAcks now takes kafka.RequireAll / RequireOne / RequireNone (was sarama.WaitForAll etc.). Publish gained an optional variadic ...Header. The removed/renamed options are WithProducerRetryBackoff, WithIdempotentProducer, WithProducerVersion, WithConsumerVersion (no kafka-go equivalent needed). See the migration note at the end of this section.

IsConnectionError is structural (like redisx's): it unwraps the error chain with errors.Is/errors.As against net.Error, io.EOF, io.ErrClosedPipe, the ECONNREFUSED/ECONNRESET/EPIPE errnos, and kafka-go's BrokerNotAvailable/LeaderNotAvailable/NetworkException/ RequestTimedOut codes.

Producer

Durable by default: RequireAll acks, MaxAttempts = 3, a 10s ack/dial/read/ write timeout, and Snappy compression. Each Publish/PublishBatch call flushes immediately in a single batched round-trip (the writer's BatchSize is pinned to

1. and returns an error only on permanent failure.

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

prod, err := kafka.NewProducer(
    []string{"localhost:19092"},
    kafka.WithClientID("stock-service"),
    kafka.WithProducerRetries(5),          // total attempts (MaxAttempts)
    kafka.WithProducerTimeout(10*time.Second),
    kafka.WithProducerCompression(kafka.CompressionSnappy), // default
    // kafka.WithRequiredAcks(kafka.RequireOne), // weaken only on purpose
)
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)

// With record headers (market-data stamps event_type/source on every message):
err = prod.Publish(ctx, "stock.quotes.realtime", []byte("AAPL"), payload,
    kafka.Header{Key: "event_type", Value: []byte("quote")},
    kafka.Header{Key: "source",     Value: []byte("alpaca")},
)

// Structured form (Headers carried as a map):
err = prod.PublishMessage(ctx, kafka.Message{
    Topic:   "stock.quotes.realtime",
    Key:     []byte("AAPL"),
    Value:   payload,
    Headers: map[string][]byte{"event_type": []byte("quote")},
})

// Batch: all messages delivered in ONE WriteMessages round-trip (all-or-nothing).
err = prod.PublishBatch(ctx, "stock.quotes.realtime", []kafka.Message{
    {Key: []byte("AAPL"), Value: p1, Headers: map[string][]byte{"event_type": []byte("quote")}},
    {Key: []byte("MSFT"), Value: p2},
})

Producer signatures & options

func NewProducer(brokers []string, opts ...ProducerOption) (*Producer, error)
func (p *Producer) Publish(ctx context.Context, topic string, key, value []byte, headers ...Header) error
func (p *Producer) PublishMessage(ctx context.Context, msg Message) error
func (p *Producer) PublishBatch(ctx context.Context, topic string, msgs []Message) error
func (p *Producer) Close() error

type Header struct { Key string; Value []byte }

WithRequiredAcks(acks RequiredAcks)        // RequireAll (default) | RequireOne | RequireNone
WithProducerRetries(n int)                 // default 3 (MaxAttempts), clamps <1 to 1
WithProducerTimeout(d time.Duration)       // default 10s (ack + net timeouts)
WithClientID(id string)
WithProducerCompression(codec Compression) // CompressionSnappy (default) | Gzip | Lz4 | Zstd | None

ConsumerGroup

NewConsumerGroup joins a group via a kafka-go Reader with GroupID + GroupTopics (multiple topics per group are supported — e.g. alert-service consuming decisions + rankings), and dispatches each record to a Handler. Run(ctx) blocks until ctx is cancelled, retrying transient fetch errors after a short backoff and shutting down gracefully on cancellation.

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", "trading.rankings"}, // topics
    handler,
    kafka.WithInitialOffset(kafka.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 OffsetPosition) // OffsetOldest (default) | OffsetNewest
WithOnError(policy OnErrorPolicy)        // MarkAndContinue (default) | Halt
WithConsumerClientID(id string)

Commit & error policy

The runner owns the commit discipline, using FetchMessage (no auto-commit)

• manual CommitMessages. After the handler returns for a message:

• Handler success → the message is committed (CommitMessages), advancing the committed offset. A restarted consumer in the same group does not reprocess committed messages.
• Handler error → behaviour depends on WithOnError:
  • MarkAndContinue (default) — logs the error and commits anyway, 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 commit (offset stays put), ends the run 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.

Migrating from v0.3.0 (sarama) to v0.4.0 (kafka-go)

• Drop the sarama import. Replace sarama.OffsetOldest/OffsetNewest in WithInitialOffset(...) with kafka.OffsetOldest / kafka.OffsetNewest.
• Acks enum. Replace WithRequiredAcks(sarama.WaitForAll) with kafka.RequireAll (and WaitForLocal→RequireOne, NoResponse→RequireNone).
• Headers (new). Publish now accepts optional ...kafka.Header; or use PublishMessage/PublishBatch with Message.Headers (map[string][]byte).
• Removed options (no kafka-go equivalent / folded in): WithProducerRetryBackoff, WithIdempotentProducer, WithProducerVersion, WithConsumerVersion. WithProducerRetries now maps to kafka-go MaxAttempts (total attempts; clamps <1 to 1).
• Compression is now selected with the neutral Compression enum (kafka.CompressionSnappy, etc.) instead of sarama.CompressionCodec.
• The Message, Handler, Producer/ConsumerGroup method shapes, the MarkAndContinue/Halt policy, and the commit semantics are otherwise unchanged.

---

clock

The time source a service reads "now" from. Calling time.Now() directly makes a service impossible to replay: every relative-duration gate in the platform (signal debounce, context staleness, earnings staleness, cache TTLs) compares now against a stored instant, so driving historical data past a wall clock makes those gates behave in ways that never happen in production.

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

clk, err := clock.FromEnv(ctx, redisClient) // System() unless CLOCK_MODE=replay
if err != nil {
    return err
}
if clk.Now().Sub(lastPublish) < debounce {
    return // suppressed
}

The default is always the real clock. Replay is opt-in via CLOCK_MODE=replay, and any other value — including a typo like raplay — resolves to real time, so a production service can never be put onto a simulated clock by accident.

Env var	Default	Purpose
CLOCK_MODE	real	replay reads simulated time; anything else is real
CLOCK_SIM_KEY	sim:clock	Redis key holding RFC3339Nano simulated time

For tests, clock.Manual(t) gives a clock you move with Set/Advance instead of sleeping.

ReplayClock never falls back to wall-clock time. A silent fallback would stamp today's date onto a replay of 2021 and corrupt the run invisibly — the worst failure mode, because the output still looks plausible. It must be primed (FromEnv does this, and errors if simulated time is unreadable), and if the source later fails it holds the last-known-good instant and logs, so a broken replay stalls visibly. Degraded() exposes that state for a health check.

---

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.