CLAUDE.md - Development Guide

Project Philosophy

Build a production-grade crypto reference data platform demonstrating staff/principal-level data engineering excellence. Prioritize simplicity, correctness, and maintainability over clever solutions.

Core Principle: Make the simple things easy and the complex things possible.

Staff-Level Standards

Architectural Thinking

Before writing code, think through:

What problem are we actually solving?
What are the constraints and tradeoffs?
What will be hard to change later? (Get these right first)
What can evolve incrementally? (Keep these simple)

Document decisions in ADRs using this format:

# ADR-NNN: Title

## Status
Accepted | Proposed | Deprecated

## Context
What forces are at play? What constraints exist?

## Decision
What did we decide? Be specific.

## Consequences
What becomes easier? What becomes harder?

## Alternatives Considered
What else did we evaluate? Why not those?

Required ADRs:

Bitemporal modeling approach (SCD Type 2 + system time)
Ingestion strategy (polling vs streaming)
Manual override workflow
API design principles
Schema evolution strategy

Code Quality Non-Negotiables

Type Safety:

# YES - Explicit types, clear contracts
def get_instrument(
    exchange: str, 
    symbol: str, 
    as_of: datetime
) -> Optional[Instrument]:
    ...

# NO - Stringly-typed, unclear contracts
def get_instrument(exchange, symbol, as_of=None):
    ...

Error Handling:

# YES - Explicit error cases, proper logging
try:
    instrument = fetch_from_exchange(exchange, symbol)
except ExchangeAPIError as e:
    logger.error("Exchange API failed", extra={
        "exchange": exchange,
        "symbol": symbol,
        "error": str(e)
    })
    raise IngestionError(f"Failed to fetch {symbol}") from e

# NO - Silent failures, generic exceptions
try:
    instrument = fetch_from_exchange(exchange, symbol)
except:
    pass

Separation of Concerns:

# YES - Clear layers, single responsibility
class BinanceClient:  # Infrastructure
    def fetch_exchange_info(self) -> dict: ...

class InstrumentParser:  # Domain logic
    def parse(self, raw: dict) -> Instrument: ...

class KafkaProducer:  # Infrastructure
    def publish(self, topic: str, event: Event): ...

# NO - God class doing everything
class BinanceIngestion:
    def fetch_parse_and_publish(self): ...

Anti-Patterns to Avoid

Over-Engineering:

❌ Building for hypothetical scale (10M instruments when you have 1k)
❌ Premature abstraction (interfaces for single implementations)
❌ Complex frameworks when stdlib/simple library works
✅ Simple solutions that scale when needed
✅ Extension points at system boundaries only
✅ Standard libraries first, dependencies when necessary

Under-Engineering:

❌ No tests ("I'll add them later")
❌ No error handling ("Happy path only")
❌ No documentation ("Code is self-documenting")
✅ Test critical paths immediately
✅ Handle errors at system boundaries
✅ Document decisions and public interfaces

Documentation Smells:

❌ Stale docs (code changed, docs didn't)
❌ Obvious docs (# Increment counter for counter += 1)
❌ Missing "why" (code shows what, docs should explain why)
✅ Living docs (updated with code changes)
✅ ADRs for decisions, docstrings for public APIs
✅ Runbooks for operational procedures

Development Workflow

1. Design First, Code Second

For each feature:

Write ADR if it involves a significant decision
Define data model (DDL/schema)
Define API contract (request/response schemas)
Write tests for critical paths
Implement
Update documentation

2. Commit Standards

Commit messages:

feat: Add bitemporal instrument table

- Implement SCD Type 2 with business/system time
- Add DBT tests for temporal consistency
- Document in ADR-001

Closes #12

Update docs in same commit:

Changed API? Update OpenAPI spec
New table? Update schema docs
New decision? Write/update ADR
Changed workflow? Update runbook

3. Testing Strategy

Test Pyramid:

     /\
    /E2E\      <- Few (happy path + critical scenarios)
   /------\
  /Integration\  <- Some (component boundaries)
 /------------\
/    Unit      \ <- Many (business logic, edge cases)

What to test:

✅ Bitemporal query logic (past/current/future states)
✅ SCD Type 2 transformations (change detection)
✅ API contracts (request validation, response schemas)
✅ Data quality rules (DBT tests)
❌ Trivial getters/setters
❌ Third-party library internals

Critical test cases:

# Test bitemporal correctness
def test_query_before_change_returns_old_spec():
    """
    Given: Instrument with tick_size change on 2024-01-15
    When: Query as_of 2024-01-14
    Then: Returns old tick_size
    """

def test_late_correction_preserves_history():
    """
    Given: Initial change record + late correction
    When: Query as_of original timestamp
    Then: Returns correct state based on valid_from
    """

Project Structure Standards

src/
├── ingestion/          # Data ingestion layer
│   ├── __init__.py
│   ├── sources/        # Exchange adapters (Binance, Kraken)
│   ├── producers.py    # Kafka producers
│   └── schemas/        # Avro/JSON schemas
├── api/                # FastAPI application
│   ├── __init__.py
│   ├── main.py         # App entrypoint
│   ├── routers/        # Route definitions
│   ├── models.py       # Pydantic models
│   ├── dependencies.py # Dependency injection
│   └── middleware/     # Auth, logging, CORS
├── common/             # Shared utilities
│   ├── config.py       # Environment config
│   ├── logging.py      # Structured logging setup
│   └── db.py           # DB connections
└── cli/                # Command-line tools

dbt/
├── models/
│   ├── bronze/         # bronze_instruments.sql
│   ├── silver/         # silver_instruments.sql (SCD Type 2)
│   └── gold/           # gold_symbology.sql
├── macros/             # scd_type2.sql, bitemporal_query.sql
├── tests/              # Data quality tests
└── docs/               # DBT docs

tests/
├── unit/               # Fast, isolated tests
├── integration/        # Cross-component tests
└── e2e/                # End-to-end scenarios

docs/
├── architecture/
│   ├── ADR-001-bitemporal-design.md
│   ├── ADR-002-ingestion-strategy.md
│   ├── SCHEMA.md       # ERD, table definitions
│   └── ARCHITECTURE.md # System overview
├── api/
│   └── openapi.yaml    # Auto-generated from FastAPI
├── runbooks/
│   ├── deployment.md
│   ├── troubleshooting.md
│   └── manual-override.md
└── development/
    ├── SETUP.md        # Local dev setup
    └── TESTING.md      # How to run tests

Naming Conventions

Python:

Modules: snake_case.py
Classes: PascalCase
Functions/variables: snake_case
Constants: SCREAMING_SNAKE_CASE
Private: _leading_underscore

SQL/DBT:

Tables: snake_case
Columns: snake_case
Layer prefix: bronze_, silver_, gold_

Kafka:

Topics: domain.entity.type (e.g., refdata.instruments.raw)
Consumer groups: service-name-consumer

Configuration Management

Environment-based config (12-factor):

# config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # Kafka
    kafka_bootstrap_servers: str
    schema_registry_url: str
    
    # MinIO/S3
    s3_endpoint_url: str
    s3_access_key: str
    s3_secret_key: str
    
    # Database
    postgres_url: str
    
    # API
    api_host: str = "0.0.0.0"
    api_port: int = 8000
    
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

settings = Settings()

NO hardcoded values:

# NO
kafka_broker = "localhost:9092"

# YES
kafka_broker = settings.kafka_bootstrap_servers

Logging Standards

Structured logging:

import structlog

logger = structlog.get_logger()

# YES - Structured, searchable
logger.info("instrument_ingested", 
    exchange="binance",
    symbol="BTCUSDT",
    tick_size=0.01,
    duration_ms=150
)

# NO - Unstructured string
logger.info(f"Ingested BTCUSDT from binance with tick_size 0.01 in 150ms")

Log levels:

DEBUG: Detailed diagnostic info (disabled in production)
INFO: Significant events (ingestion started, API request)
WARNING: Degraded state (API slow, retry attempted)
ERROR: Failure requiring attention (ingestion failed)
CRITICAL: System-level failure (database down)

API Design Principles

RESTful conventions:

GET    /v1/instruments                    # List instruments
GET    /v1/instruments/{id}               # Get specific instrument
POST   /v1/instruments                    # Create (manual override)
GET    /v1/instruments/{id}/history       # Get audit trail
GET    /v1/symbology/{canonical_id}       # Symbology lookup
GET    /v1/health                         # Health check

Consistent error responses:

{
  "error": {
    "code": "INSTRUMENT_NOT_FOUND",
    "message": "Instrument binance/BTCUSDT not found",
    "details": {
      "exchange": "binance",
      "symbol": "BTCUSDT",
      "as_of": "2024-01-15T10:00:00Z"
    },
    "request_id": "req_abc123"
  }
}

Query parameters for filtering:

GET /v1/instruments?exchange=binance&asset_class=spot&as_of=2024-01-15T10:00:00Z

Pagination for large results:

GET /v1/instruments?limit=100&offset=0

Response:
{
  "data": [...],
  "pagination": {
    "total": 1543,
    "limit": 100,
    "offset": 0,
    "next": "/v1/instruments?limit=100&offset=100"
  }
}

Data Quality Standards

DBT tests for every model:

# models/silver/silver_instruments.yml
version: 2

models:
  - name: silver_instruments
    description: Bitemporal instrument specifications
    columns:
      - name: instrument_id
        tests:
          - unique
          - not_null
      
      - name: tick_size
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
              inclusive: false
      
      - name: valid_from
        tests:
          - not_null
      
      - name: valid_to
        tests:
          - dbt_utils.expression_is_true:
              expression: "valid_to IS NULL OR valid_to > valid_from"

Freshness checks:

sources:
  - name: bronze
    tables:
      - name: bronze_instruments
        freshness:
          warn_after: {count: 6, period: hour}
          error_after: {count: 12, period: hour}

Observability

Required metrics:

Ingestion: messages/sec, lag, error rate
API: request rate, latency (p50/p95/p99), error rate
Data quality: test failures, freshness violations
Infrastructure: CPU, memory, disk I/O

Health check endpoint:

@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "version": "1.0.0",
        "checks": {
            "database": "ok",
            "kafka": "ok",
            "s3": "ok"
        }
    }

Documentation Workflow

For every PR:

Update relevant ADR if decision changes
Update API docs (OpenAPI spec) if endpoints change
Update schema docs if data model changes
Update runbook if operational procedures change
Add inline comments for complex logic only

When to write docs:

ADR: When making architectural decisions
Docstring: For all public functions/classes
Comment: For non-obvious logic (why, not what)
Runbook: For operational procedures
README: For setup and getting started

Documentation checklist before merge:

ADR written/updated if architectural decision made
API docs reflect endpoint changes
Schema docs reflect table changes
Tests document expected behavior
Runbook updated if operational impact

Code Review Principles

What to look for:

✅ Are decisions documented (ADR)?
✅ Are error cases handled?
✅ Are there tests for critical paths?
✅ Is the code self-explanatory?
✅ Is documentation updated?
❌ Don't nitpick formatting (use automated tools)
❌ Don't bikeshed (focus on substance)

Review checklist:

Incremental Development

Build in thin vertical slices:

Slice 1: Bronze ingestion (end-to-end spike)

Binance client fetches /exchangeInfo
Publishes to Kafka topic
Writes to Bronze Iceberg table
Basic health check
Document in ADR-002 (ingestion strategy)

Slice 2: Silver transformation

DBT model for SCD Type 2
Parse JSON to relational schema
Data quality tests
Update schema docs

Slice 3: API query layer

FastAPI endpoint for current state
Pydantic models
OpenAPI spec
Integration test

Slice 4: Bitemporal queries

Add as_of parameter
Implement temporal logic
Test historical queries
Document in ADR-001 (bitemporal design)

Each slice is deployable and adds value.

Principal-Level Mindset

Think in systems, not features:

How does this fit into the broader platform?
What operational burden does this create?
What will future maintainers need to know?
How do we know if this is working in production?

Design for evolution:

Make common things easy (simple API)
Make uncommon things possible (extension points)
Provide clear upgrade paths (versioning)
Document constraints and assumptions (ADRs)

Optimize for maintenance:

Code is read 10x more than written
Clear is better than clever
Explicit is better than implicit
Simple is better than complex

Own the outcomes:

It's not done until it's documented
It's not tested until edge cases are covered
It's not production-ready until it's observable
It's not maintainable until someone else understands it

Final Checklist

Before considering a component "complete":

Remember: Staff/principal engineers are measured not by lines of code, but by enabling teams to ship high-quality systems efficiently. Focus on leverage—the work that multiplies the effectiveness of others.

FilesExpand file tree

CLAUDE.md

Latest commit

History