FinPull - Real-time Market Data Collector

Clean Architecture + Dependency Injection (Wire) implementation for collecting real-time market data from Finnhub and streaming to Kafka/ClickHouse.

Architecture

pkg/                          # Shared infrastructure layer
├── server/app.go              → Application lifecycle (Start/Stop/Health/Ready)
├── clickhouse/client.go       → Connection pool manager + schema initialization
├── kafka/client.go            → Writer manager + batch/compression config
├── config/config.go           → YAML config loader + validation
└── metrics/prometheus.go      → Prometheus metrics recorder

internal/
├── domain/
│   ├── models/trade.go        → Business entities
│   └── repository/            → Port interfaces (abstractions)
│       └── interfaces.go      → MarketStream, Publisher, Storage, Metrics
├── repository/                → Adapters (interface implementations)
│   ├── clickhouse/storage.go  → ClickHouse storage adapter
│   └── kafka/publisher.go     → Kafka publisher adapter
├── service/
│   └── finnhub/client.go      → Finnhub WebSocket client (third-party)
├── usecase/                   → Business logic
│   ├── trade_processor.go     → Process & route trades to backend
│   └── trade_collector.go     → Collect trades from market stream
└── di/                        → Dependency injection
    ├── providers.go           → Provider functions for Wire
    └── wire.go                → Wire injector definition

cmd/app/main.go                → Entry point (3-step: load config → wire DI → run)

Key Principles

1. Clean Architecture

• Domain layer: Pure business logic, no external dependencies
• Use case layer: Application-specific business rules
• Interface adapters: Convert data between domain and external systems
• Infrastructure: Frameworks, tools, drivers (ClickHouse, Kafka, HTTP)

2. Dependency Injection with Wire

• All dependencies injected via Wire
• No global state or singletons
• Easy to test and mock

3. Separation of Concerns

• pkg/: Infrastructure setup (connections, configs) - REUSABLE
• internal/: Application logic (domain, usecases, adapters) - BUSINESS SPECIFIC
• cmd/: Entry points - MINIMAL (3 steps)

4. Connection Management

• Connections created ONCE in pkg/
• Repositories receive connections, never create them
• Cleanup handled centrally

Quick Start

Build

make build
# or
go build -o bin/finpull ./cmd/app

Run

# Development
make run
# or
./bin/finpull -config config/dev.yaml

# Production
./bin/finpull -config config/production.yaml

Configuration

config/dev.yaml:

environment: development
backend:
  type: kafka  # or clickhouse
finnhub:
  api_key: ${FINNHUB_API_KEY}
  symbols:
    - BINANCE:BTCUSDT
kafka:
  brokers:
    - kafka-headless:9092
  topic: finpull

Environment variables override YAML:

export FINNHUB_API_KEY=your_key
export BACKEND=clickhouse
./bin/finpull

Development

Install Tools

make install-tools

Installs:

• Wire (dependency injection code generator)
• golangci-lint (linter)

Generate Wire Code

make wire

Generates internal/di/wire_gen.go.

Lint

make lint

Enforces:

• Cyclomatic complexity < 15
• Cognitive complexity < 15
• Function length < 100 lines
• Error handling
• Code style

Test

make test

Docker

Start Infrastructure

make docker-up

Starts:

• ClickHouse (:8123, :9000)
• Kafka (:9092)
• Zookeeper
• Postgres (:5432)
• Redis (:6379)
• Superset (:8088)
• Prometheus (:9090)
• Grafana (:3000)

Stop Infrastructure

make docker-down

Endpoints

• Health: GET /health - Infrastructure health (ClickHouse + Kafka)
• Ready: GET /ready - Application readiness (Collector connected)
• Metrics: GET /metrics - Prometheus metrics

Metrics

• finpull_messages_sent_total{backend,symbol} - Total messages sent
• finpull_errors_total{type} - Total errors
• finpull_last_price{symbol} - Last recorded price
• finpull_operation_duration_seconds{operation} - Operation latency

Project Structure Best Practices

What Goes Where?

pkg/ - Infrastructure utilities (reusable across projects):

• Connection managers (DB, Kafka, Redis)
• Configuration loaders
• HTTP servers
• Metrics recorders
• Logging utilities

internal/ - Application-specific logic:

• Domain models and interfaces
• Business logic (use cases)
• Interface adapters (repositories, services)
• Dependency injection setup

cmd/ - Executable entry points:

• Parse flags
• Load config
• Wire dependencies
• Run application
• That's it. No business logic.

Anti-patterns to Avoid

❌ DO NOT create connections in repositories
✅ DO inject connections from pkg/

❌ DO NOT put business logic in cmd/
✅ DO keep cmd/ minimal (3 steps)

❌ DO NOT import internal/ from pkg/
✅ DO keep pkg/ independent and reusable

❌ DO NOT use global variables
✅ DO inject dependencies via Wire

License

MIT

Contributing

1. Follow Clean Architecture principles
2. Run make lint before commit
3. Keep cyclomatic complexity < 15
4. Add tests for new features
5. Update docs

---

Built with: Go 1.24 | Wire | ClickHouse | Kafka | Prometheus | Grafana