Structured Logging for Financial APIs: From print() to structlog

---

The 3 AM Wake-Up Call

Your phone buzzes at 3:14 AM. The trading system reported an anomaly during the Tokyo session. You SSH into the server, scroll through logs that look like this:

2026-04-01 03:14:22 Processing order
2026-04-01 03:14:22 Order validated
2026-04-01 03:14:22 ERROR Something went wrong
2026-04-01 03:14:23 Retrying
2026-04-01 03:14:25 Failed again

Which order? What validation failed? Which retry? You spend 45 minutes reconstructing the sequence from fragmented timestamps before you can even begin to debug.

This scenario is not hypothetical. It is the predictable outcome of writing logs the way we were taught: with print() statements and string interpolation. In high-frequency financial systems where millisecond latency matters and order integrity is non-negotiable, unstructured logging is a liability that delays incident response and obscures the audit trail.

This article is a practical guide to structured logging for Python-based financial APIs. We will move from naive print() patterns through production-grade structlog configurations, JSON serialization, and ELK stack integration. Every code example is production-ready and designed for the specific demands of financial data infrastructure.

---

The Problem with Unstructured Logging

What "Naive" Logging Looks Like

The typical entry point into logging is the print() statement:

def process_order(order_id, symbol, quantity, price):
    print(f"[{datetime.now()}] Processing order {order_id}: {symbol} x {quantity} @ {price}")
    
    if quantity <= 0:
        print(f"[ERROR] Invalid quantity {quantity} for order {order_id}")
        return None
    
    print(f"[{datetime.now()}] Order {order_id} validated, submitting to exchange")
    result = submit_to_exchange(symbol, quantity, price)
    
    print(f"[{datetime.now()}] Order {order_id} result: {result}")
    return result

This code is readable. It is also nearly useless for production debugging.

The Four Failures of String Interpolation Logging

Failure Mode	Symptom	Impact
Timestamp inconsistency	Manual datetime formatting drifts; log aggregation breaks	Cannot correlate events across services
No severity hierarchy	[ERROR] and [DEBUG] require regex parsing	Alerting systems cannot distinguish critical from noise
Flat strings, no context	order_id, symbol, quantity are buried in text	Cannot filter or search programmatically
No structured metadata	Cannot compute metrics (error rate, latency percentiles) from logs	Observability blind spots

Financial APIs have additional requirements that naive logging cannot satisfy:

1. Audit compliance: Regulators may require immutable, queryable records of every order event.
2. Cross-service correlation: An order may flow through API gateway → validation service → risk engine → exchange adapter. Without a correlation ID, tracing this flow is archaeological work.
3. Sensitive data handling: PII (client IDs, account numbers) must be redacted before logging.
4. High throughput: A busy trading API may emit thousands of log entries per second. String concatenation per log line is CPU-expensive.

---

What Structured Logging Actually Means

The Core Principle

Structured logging is the practice of emitting log entries as key-value pairs rather than formatted strings. Instead of:

2026-04-01 15:30:45 ERROR Failed to submit order AAPL-20260401-001 for client C-88291: insufficient margin

We emit:

{
  "timestamp": "2026-04-01T15:30:45.123Z",
  "level": "error",
  "event": "order_submission_failed",
  "order_id": "AAPL-20260401-001",
  "client_id": "C-88291",
  "symbol": "AAPL",
  "reason": "insufficient_margin",
  "available_margin": 12500.00,
  "required_margin": 18750.00
}

Every field is queryable. Every entry is typed. The same event from different services can be joined on order_id. An ELK query for reason:insufficient_margin AND symbol:AAPL returns precise results in milliseconds.

Why JSON?

JSON is the lingua franca of log aggregation pipelines:

• Parsable: Every major log shipper (Filebeat, Fluentd, Logstash) has native JSON parsers.
• Extensible: Add new fields without breaking existing queries.
• Typed: Numbers remain numbers; booleans remain booleans. This matters for range queries and aggregations in Elasticsearch.

The tradeoff is human readability in development. structlog addresses this with dual rendering: JSON for production, colored console output for development.

---

Introducing structlog

structlog is a Python library that transforms logging from a string-output problem into a data-output problem. It provides:

• Bound loggers: Attach context that persists across function calls.
• Processors: Transform log entries (add timestamps, redact PII, inject correlation IDs).
• Renderers: Output to JSON, console, or any custom format.
• Standard library compatibility: Can route through Python's built-in logging module.

Installation

pip install structlog

Basic Usage

import structlog

# Initialize structlog
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer()
    ],
    wrapper_class=structlog.stdlib.BoundLogger,
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)

log = structlog.get_logger()

log.info(
    "order_submitted",
    order_id="AAPL-20260401-001",
    symbol="AAPL",
    quantity=100,
    price=185.50,
    client_id="C-88291"
)

Output:

{
  "order_id": "AAPL-20260401-001",
  "symbol": "AAPL",
  "quantity": 100,
  "price": 185.5,
  "client_id": "C-88291",
  "event": "order_submitted",
  "level": "info",
  "timestamp": "2026-04-01T15:30:45.123456Z"
}

---

Production-Grade Configuration for Financial APIs

The Financial API Processor Chain

A production financial API requires a more sophisticated processor chain than the basic example. Here is a configuration designed for high-throughput trading systems:

import structlog
import logging
import sys
import os
from datetime import datetime
from typing import Any

# Detect environment
ENVIRONMENT = os.environ.get("ENVIRONMENT", "development")
IS_PRODUCTION = ENVIRONMENT == "production"


def add_service_context(logger, method_name, event_dict):
    """Inject service-level metadata into every log entry."""
    event_dict["service"] = os.environ.get("SERVICE_NAME", "unknown")
    event_dict["version"] = os.environ.get("SERVICE_VERSION", "0.0.0")
    event_dict["environment"] = ENVIRONMENT
    return event_dict


def redact_sensitive_fields(logger, method_name, event_dict):
    """Redact PII and financial identifiers before logging."""
    sensitive_keys = {"client_id", "account_number", "ssn", "credit_card", "api_key"}
    
    for key in sensitive_keys:
        if key in event_dict:
            original = str(event_dict[key])
            # Preserve first and last 4 characters for traceability
            if len(original) > 8:
                event_dict[key] = f"{original[:4]}...{original[-4:]}"
            else:
                event_dict[key] = "***REDACTED***"
    
    return event_dict


def add_correlation_id(logger, method_name, event_dict):
    """Ensure every log entry has a correlation ID for request tracing."""
    correlation_id = event_dict.pop("correlation_id", None)
    if correlation_id is None:
        correlation_id = os.environ.get("_CORRELATION_ID", "no-correlation-id")
    event_dict["correlation_id"] = correlation_id
    return event_dict


def ensure_order_context(logger, method_name, event_dict):
    """Attach order-specific context if present in the call stack."""
    # Check for order context set via bound logger
    if hasattr(logger, "_context") and "order_id" in logger._context:
        for key in ["order_id", "symbol", "side", "quantity"]:
            if key not in event_dict and key in logger._context:
                event_dict[key] = logger._context[key]
    return event_dict


# Define processors based on environment
if IS_PRODUCTION:
    processors = [
        structlog.stdlib.filter_by_level,
        add_service_context,
        add_correlation_id,
        ensure_order_context,
        redact_sensitive_fields,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer(),
    ]
else:
    # Development: human-readable console output
    processors = [
        structlog.stdlib.filter_by_level,
        add_service_context,
        add_correlation_id,
        structlog.stdlib.add_log_level,
        structlog.dev.ConsoleRenderer(colors=True),
    ]


structlog.configure(
    processors=processors,
    wrapper_class=structlog.stdlib.BoundLogger,
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)


def get_logger(name: str = None) -> structlog.stdlib.BoundLogger:
    """Factory function for getting configured loggers."""
    return structlog.get_logger(name)

Creating Bound Loggers for Order Flow

The bound logger pattern is essential for financial APIs where an order traverses multiple services:

def process_order(order_id: str, symbol: str, quantity: int, side: str, client_id: str):
    """
    Process an order through the trading pipeline.
    
    Demonstrates bound logger pattern for maintaining context
    across service boundaries.
    """
    log = get_logger("order_processor").bind(
        order_id=order_id,
        symbol=symbol,
        side=side,
        client_id=client_id  # Will be redacted in production
    )
    
    log.info("order_processing_started", quantity=quantity)
    
    try:
        # Step 1: Risk validation
        risk_log = log.bind(phase="risk_validation")
        risk_log.debug("evaluating_risk_parameters")
        
        risk_result = validate_risk(order_id, symbol, quantity, client_id)
        
        if not risk_result.approved:
            risk_log.warning(
                "risk_validation_rejected",
                reason=risk_result.reason,
                margin_used=risk_result.margin_used,
                margin_available=risk_result.margin_available
            )
            return {"status": "rejected", "reason": risk_result.reason}
        
        risk_log.info("risk_validation_passed")
        
        # Step 2: Exchange submission
        exchange_log = log.bind(phase="exchange_submission")
        exchange_log.info("submitting_to_exchange")
        
        exchange_result = submit_to_exchange(symbol, quantity, side)
        
        if exchange_result.status == "filled":
            exchange_log.info(
                "order_filled",
                fill_price=exchange_result.fill_price,
                fill_quantity=exchange_result.fill_quantity,
                latency_ms=exchange_result.latency_ms
            )
        else:
            exchange_log.error(
                "exchange_rejected",
                rejection_code=exchange_result.code,
                rejection_reason=exchange_result.reason
            )
        
        return {
            "status": exchange_result.status,
            "fill_price": exchange_result.fill_price,
            "exchange_order_id": exchange_result.exchange_order_id
        }
        
    except Exception as exc:
        log.error(
            "order_processing_failed",
            exception_type=type(exc).__name__,
            exception_message=str(exc),
            # Include traceback via structlog's exception handling
        )
        raise

Middleware for HTTP Request Correlation

For REST APIs, inject correlation IDs at the request boundary:

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
import uuid


class StructuredLoggingMiddleware(BaseHTTPMiddleware):
    """
    Middleware that:
    1. Generates a correlation ID for each request
    2. Injects it into the log context
    3. Returns it in response headers for client-side tracing
    """
    
    async def dispatch(self, request: Request, call_next):
        # Extract or generate correlation ID
        correlation_id = request.headers.get("X-Correlation-ID")
        if not correlation_id:
            correlation_id = str(uuid.uuid4())
        
        # Store in environment for subprocess access
        os.environ["_CORRELATION_ID"] = correlation_id
        
        # Bind to logger
        log = get_logger("http").bind(
            correlation_id=correlation_id,
            method=request.method,
            path=request.url.path,
            client_ip=self._extract_client_ip(request)
        )
        
        log.info("request_received")
        
        start_time = time.time()
        
        try:
            response = await call_next(request)
            
            duration_ms = (time.time() - start_time) * 1000
            
            log.info(
                "request_completed",
                status_code=response.status_code,
                duration_ms=round(duration_ms, 2)
            )
            
            # Include correlation ID in response for client tracing
            response.headers["X-Correlation-ID"] = correlation_id
            
            return response
            
        except Exception as exc:
            duration_ms = (time.time() - start_time) * 1000
            
            log.error(
                "request_failed",
                exception_type=type(exc).__name__,
                duration_ms=round(duration_ms, 2)
            )
            raise
    
    @staticmethod
    def _extract_client_ip(request: Request) -> str:
        forwarded = request.headers.get("X-Forwarded-For")
        if forwarded:
            return forwarded.split(",")[0].strip()
        return request.client.host if request.client else "unknown"

---

ELK Stack Integration

Architecture Overview

The ELK stack (Elasticsearch, Logstash, Kibana) is the standard for centralized log analysis in distributed systems. Here is how structured logs flow through the pipeline:

Application → structlog (JSON) → stdout → Filebeat → Logstash → Elasticsearch → Kibana

Filebeat Configuration

# filebeat.yml
filebeat.inputs:
  - type: container
    paths:
      - /var/log/containers/*.log
    json.keys_under_root: true
    json.add_error_key: true
    json.message_key: message
    fields:
      service: tickdb-api
      environment: production
    fields_under_root: true

processors:
  - add_host_metadata:
      when.not.contains.tags: forwarded
  - add_cloud_metadata: ~
  - add_docker_metadata: ~

output.elasticsearch:
  hosts: ["elasticsearch:9200"]
  index: "tickdb-logs-%{+yyyy.MM.dd}"

setup.kibana:
  host: "kibana:5601"

setup.ilm.enabled: true
setup.ilm.rollover_alias: "tickdb-logs"
setup.ilm.pattern: "{now/d}-000001"
setup.ilm.policy_name: "tickdb-logs-policy"

Elasticsearch Index Template

Define field mappings for financial API logs:

{
  "index_patterns": ["tickdb-logs-*"],
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "index.lifecycle.name": "tickdb-logs-policy",
      "index.lifecycle.rollover_alias": "tickdb-logs"
    },
    "mappings": {
      "properties": {
        "timestamp": { "type": "date" },
        "level": { "type": "keyword" },
        "event": { "type": "keyword" },
        "order_id": { "type": "keyword" },
        "symbol": { "type": "keyword" },
        "quantity": { "type": "long" },
        "price": { "type": "double" },
        "side": { "type": "keyword" },
        "client_id": { "type": "keyword" },
        "correlation_id": { "type": "keyword" },
        "service": { "type": "keyword" },
        "environment": { "type": "keyword" },
        "duration_ms": { "type": "float" },
        "status_code": { "type": "integer" },
        "reason": { "type": "text", "fields": { "keyword": { "type": "keyword" } } },
        "exception_type": { "type": "keyword" },
        "exception_message": { "type": "text" }
      }
    }
  }
}

Kibana Saved Searches

Create saved searches for common debugging scenarios:

1. Order Tracking by Correlation ID

correlation_id: "a1b2c3d4-e5f6-7890-abcd-ef1234567890"

2. Failed Orders in Time Window

level: error AND event: order_* AND @timestamp: [now-1h TO now]

3. High Latency Requests

duration_ms: > 1000 AND event: request_completed

4. Symbol-Specific Analysis

symbol: AAPL AND @timestamp: [now-24h TO now]
| stats avg(duration_ms), count() by event

---

Advanced Patterns for Financial Systems

Distributed Tracing Integration

For microservices architectures, propagate trace context:

from opentelemetry import trace
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator

propagator = TraceContextTextMapPropagator()


def inject_trace_context(logger):
    """Extract trace context from OpenTelemetry and inject into log context."""
    carrier = {}
    propagator.inject(carrier)
    
    span = trace.get_current_span()
    if span and span.get_span_context().is_valid:
        span_context = span.get_span_context()
        carrier["trace_id"] = format(span_context.trace_id, "032x")
        carrier["span_id"] = format(span_context.span_id, "016x")
    
    return logger.bind(**carrier)

Performance Budget Monitoring

Financial APIs have strict latency SLAs. Log latency with percentile context:

import time
from functools import wraps
from typing import Callable
import asyncio


def log_latency(logger, operation: str, threshold_ms: float = 100):
    """Decorator to log operation latency and alert on threshold breaches."""
    def decorator(func: Callable):
        @wraps(func)
        async def async_wrapper(*args, **kwargs):
            start = time.perf_counter()
            try:
                result = await func(*args, **kwargs)
                return result
            finally:
                duration_ms = (time.perf_counter() - start) * 1000
                
                log_event = logger.info if duration_ms < threshold_ms else logger.warning
                log_event(
                    "operation_completed",
                    operation=operation,
                    duration_ms=round(duration_ms, 2),
                    threshold_ms=threshold_ms,
                    within_threshold=duration_ms < threshold_ms
                )
        
        @wraps(func)
        def sync_wrapper(*args, **kwargs):
            start = time.perf_counter()
            try:
                result = func(*args, **kwargs)
                return result
            finally:
                duration_ms = (time.perf_counter() - start) * 1000
                
                log_event = logger.info if duration_ms < threshold_ms else logger.warning
                log_event(
                    "operation_completed",
                    operation=operation,
                    duration_ms=round(duration_ms, 2),
                    threshold_ms=threshold_ms,
                    within_threshold=duration_ms < threshold_ms
                )
        
        if asyncio.iscoroutinefunction(func):
            return async_wrapper
        return sync_wrapper
    
    return decorator


# Usage
log = get_logger("market_data")

@log_latency(log, "fetch_order_book", threshold_ms=50)
def fetch_order_book(symbol: str, depth: int = 10):
    """Fetch order book from exchange with latency monitoring."""
    # Implementation
    pass

Log Sampling for High-Volume Scenarios

During extreme market volatility, log volume can overwhelm ingestion pipelines. Implement probabilistic sampling:

import random
from typing import Callable, TypeVar

F = TypeVar('F')


def sample_logs(sample_rate: float = 1.0, sample_key: str = "order_id"):
    """
    Decorator that probabilistically samples log entries.
    
    Ensures all entries for a given order_id are either
    fully sampled or fully dropped (consistent sampling).
    """
    def decorator(func: Callable[F]) -> Callable[F]:
        sampled_orders = set()
        
        def should_sample(order_id: str) -> bool:
            if order_id not in sampled_orders:
                if random.random() < sample_rate:
                    sampled_orders.add(order_id)
                    return True
                return False
            return True
        
        @wraps(func)
        def wrapper(*args, **kwargs):
            order_id = kwargs.get(sample_key) or (args[0] if args else None)
            
            if order_id and not should_sample(order_id):
                # Log sampling decision but skip full instrumentation
                get_logger(func.__module__).debug(
                    "log_sampled",
                    function=func.__name__,
                    sample_rate=sample_rate
                )
                return func(*args, **kwargs)
            
            return func(*args, **kwargs)
        
        return wrapper
    return decorator

---

Security Considerations

What Never Goes in Logs

Category	Examples	Why
Credentials	API keys, passwords, tokens	Credential stuffing attacks
PII	SSN, passport numbers, full names	GDPR, CCPA violations
Financial data	Full credit card numbers, bank account details	PCI-DSS violations
Authentication tokens	JWT contents, session cookies	Session hijacking

The Redaction Processor in Detail

import re
from typing import Any


class FinancialDataRedactor:
    """
    Production-grade redaction for financial API logs.
    
    Handles common patterns found in trading systems.
    """
    
    # Patterns for financial identifiers
    PATTERNS = {
        "api_key": re.compile(r'(api[_-]?key["\']?\s*[:=]\s*["\']?)([a-zA-Z0-9_-]{8,})'),
        "jwt": re.compile(r'(Bearer\s+)([a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]+)'),
        "credit_card": re.compile(r'\b(\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4})\b'),
        "account_number": re.compile(r'(account[_-]?num(ber)?["\']?\s*[:=]\s*["\']?)(\d{6,})', re.IGNORECASE),
    }
    
    # Fields to redact entirely
    FULL_REDACT_FIELDS = {
        "password", "secret", "token", "authorization",
        "credit_card", "cvv", "ssn", "api_secret"
    }
    
    # Fields to partially redact (first/last 4)
    PARTIAL_REDACT_FIELDS = {
        "client_id", "account_id", "order_id", "user_id"
    }
    
    @classmethod
    def redact_string(cls, value: str) -> str:
        """Apply regex-based redaction to a string value."""
        result = value
        
        # Redact API keys
        result = cls.PATTERNS["api_key"].sub(r'\1[REDACTED]', result)
        
        # Redact JWTs
        result = cls.PATTERNS["jwt"].sub(r'\1[REDACTED_JWT]', result)
        
        # Redact credit card numbers
        result = cls.PATTERNS["credit_card"].sub('[REDACTED_CC]', result)
        
        # Redact account numbers
        result = cls.PATTERNS["account_number"].sub(r'\1[REDACTED]', result, flags=re.IGNORECASE)
        
        return result
    
    @classmethod
    def redact_dict(cls, data: dict) -> dict:
        """Recursively redact sensitive fields from a dictionary."""
        if not isinstance(data, dict):
            return data
        
        result = {}
        for key, value in data.items():
            key_lower = key.lower()
            
            if key_lower in cls.FULL_REDACT_FIELDS:
                result[key] = "[REDACTED]"
            elif key_lower in cls.PARTIAL_REDACT_FIELDS and isinstance(value, str):
                result[key] = cls._partial_redact(value)
            elif isinstance(value, str):
                redacted = cls.redact_string(value)
                result[key] = redacted
            elif isinstance(value, dict):
                result[key] = cls.redact_dict(value)
            elif isinstance(value, list):
                result[key] = [
                    cls.redact_dict(item) if isinstance(item, dict) 
                    else cls.redact_string(str(item)) if isinstance(item, str)
                    else item
                    for item in value
                ]
            else:
                result[key] = value
        
        return result
    
    @staticmethod
    def _partial_redact(value: str) -> str:
        """Partially redact a string value, keeping first/last 4 chars."""
        if len(value) <= 8:
            return "[REDACTED]"
        return f"{value[:4]}...{value[-4:]}"

---

Validation and Testing

Log Output Testing

Verify that your logging configuration produces valid, queryable output:

import pytest
import json
from io import StringIO


@pytest.fixture
def log_output():
    """Capture structured log output as JSON lines."""
    stream = StringIO()
    
    structlog.configure(
        processors=[
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.JSONRenderer(),
        ],
        wrapper_class=structlog.stdlib.BoundLogger,
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(stream=stream),
        cache_logger_on_first_use=False,
    )
    
    yield stream
    stream.close()


def test_order_log_contains_required_fields(log_output):
    """Verify order logs contain all fields required for ELK indexing."""
    log = structlog.get_logger()
    
    log.info(
        "order_submitted",
        order_id="TEST-001",
        symbol="AAPL",
        quantity=100,
        price=185.50,
        side="buy"
    )
    
    log_output.seek(0)
    log_entry = json.loads(log_output.readline())
    
    # Verify required fields exist
    assert "level" in log_entry
    assert "timestamp" in log_entry
    assert log_entry["order_id"] == "TEST-001"
    assert log_entry["symbol"] == "AAPL"
    assert log_entry["quantity"] == 100
    assert log_entry["price"] == 185.50


def test_sensitive_data_is_redacted(log_output):
    """Verify sensitive fields are redacted in production output."""
    log = structlog.get_logger()
    
    log.info(
        "client_authenticated",
        client_id="C-88291",
        api_key="sk_live_abcdef123456789",
        event="authentication"
    )
    
    log_output.seek(0)
    log_entry = json.loads(log_output.readline())
    
    # API key should be redacted
    assert "sk_live" not in json.dumps(log_entry)
    # client_id partial redaction (first 4 chars preserved)
    assert log_entry["client_id"] == "C-88...8291"


def test_latency_is_logged_as_float():
    """Verify latency measurements are numeric for aggregation."""
    log = get_logger("test")
    
    # ... emit log with duration_ms ...
    
    log_output.seek(0)
    log_entry = json.loads(log_output.readline())
    
    assert isinstance(log_entry.get("duration_ms"), (int, float))

---

Putting It Together: A Complete Example

Here is the full implementation pattern for a financial API endpoint:

"""
TickDB Market Data API - Structured Logging Implementation
https://tickdb.ai
"""

import structlog
import os
import time
from functools import wraps
from typing import Any, Callable
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
import uvicorn

# ============================================================
# CONFIGURATION
# ============================================================

ENVIRONMENT = os.environ.get("ENVIRONMENT", "development")
SERVICE_NAME = os.environ.get("SERVICE_NAME", "tickdb-market-data")
SERVICE_VERSION = os.environ.get("SERVICE_VERSION", "1.0.0")

# ============================================================
# STRUCTURED LOGGING SETUP
# ============================================================

def configure_logging():
    """Configure structlog for production use."""
    
    shared_processors = [
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
    ]
    
    if ENVIRONMENT == "production":
        # Production: JSON output for ELK
        structlog.configure(
            processors=shared_processors + [
                _add_service_context,
                _add_correlation_id,
                _redact_sensitive_fields,
                structlog.processors.JSONRenderer()
            ],
            wrapper_class=structlog.stdlib.BoundLogger,
            context_class=dict,
            logger_factory=structlog.stdlib.LoggerFactory(),
            cache_logger_on_first_use=True,
        )
    else:
        # Development: colored console output
        structlog.configure(
            processors=shared_processors + [
                _add_service_context,
                _add_correlation_id,
                structlog.dev.ConsoleRenderer(colors=True)
            ],
            wrapper_class=structlog.stdlib.BoundLogger,
            context_class=dict,
            logger_factory=structlog.stdlib.LoggerFactory(),
            cache_logger_on_first_use=True,
        )


def _add_service_context(logger, method_name, event_dict):
    event_dict["service"] = SERVICE_NAME
    event_dict["version"] = SERVICE_VERSION
    event_dict["environment"] = ENVIRONMENT
    return event_dict


def _add_correlation_id(logger, method_name, event_dict):
    correlation_id = os.environ.get("_CORRELATION_ID", "no-correlation")
    event_dict["correlation_id"] = correlation_id
    return event_dict


def _redact_sensitive_fields(logger, method_name, event_dict):
    REDACTED_FIELDS = {"api_key", "password", "secret", "token"}
    
    for field in REDACTED_FIELDS:
        if field in event_dict:
            event_dict[field] = "[REDACTED]"
    
    # Partial redaction for client identifiers
    PARTIAL_FIELDS = {"client_id", "account_id"}
    for field in PARTIAL_FIELDS:
        if field in event_dict and isinstance(event_dict[field], str):
            val = event_dict[field]
            if len(val) > 8:
                event_dict[field] = f"{val[:4]}...{val[-4:]}"
    
    return event_dict


# Initialize logging
configure_logging()
log = structlog.get_logger()


# ============================================================
# MIDDLEWARE
# ============================================================

class CorrelationMiddleware:
    """Inject correlation ID into each request context."""
    
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            request = Request(scope)
            correlation_id = request.headers.get(
                "X-Correlation-ID", 
                request.headers.get("X-Request-ID", f"req-{int(time.time()*1000)}")
            )
            os.environ["_CORRELATION_ID"] = correlation_id
            
            # Log request
            log.info(
                "http_request_started",
                method=request.method,
                path=request.url.path,
                correlation_id=correlation_id
            )
        
        await self.app(scope, receive, send)


# ============================================================
# API ENDPOINTS
# ============================================================

app = Starlette(debug=ENVIRONMENT != "production", middleware=[CorrelationMiddleware])


@app.route("/v1/market/kline")
async def get_kline(request: Request):
    """
    Get OHLCV kline data for a symbol.
    
    Query parameters:
        symbol: Trading symbol (e.g., AAPL.US)
        interval: Kline interval (1m, 5m, 1h, 1d)
        limit: Number of klines to return (max 1000)
    """
    symbol = request.query_params.get("symbol")
    interval = request.query_params.get("interval", "1h")
    limit = min(int(request.query_params.get("limit", 100)), 1000)
    
    request_log = log.bind(symbol=symbol, interval=interval, limit=limit)
    request_log.info("kline_request_received")
    
    start_time = time.perf_counter()
    
    try:
        # Validate symbol
        if not symbol:
            request_log.warning("kline_request_invalid", reason="missing_symbol")
            return JSONResponse(
                {"error": "symbol parameter is required", "code": "INVALID_PARAMETER"},
                status_code=400
            )
        
        # Fetch data (placeholder for actual TickDB API call)
        # In production, this would use the TickDB SDK
        data = await fetch_kline_data(symbol, interval, limit)
        
        duration_ms = (time.perf_counter() - start_time) * 1000
        request_log.info(
            "kline_request_completed",
            duration_ms=round(duration_ms, 2),
            record_count=len(data.get("klines", []))
        )
        
        return JSONResponse(data)
        
    except Exception as exc:
        duration_ms = (time.perf_counter() - start_time) * 1000
        request_log.error(
            "kline_request_failed",
            duration_ms=round(duration_ms, 2),
            error_type=type(exc).__name__,
            error_message=str(exc)
        )
        return JSONResponse(
            {"error": "Internal server error", "code": "INTERNAL_ERROR"},
            status_code=500
        )


async def fetch_kline_data(symbol: str, interval: str, limit: int) -> dict:
    """
    Fetch kline data from TickDB.
    
    This is a placeholder implementation.
    Production code would use the TickDB Python SDK:
    
        from tickdb import TickDB
        
        client = TickDB(api_key=os.environ["TICKDB_API_KEY"])
        data = client.market.kline.get(
            symbol=symbol,
            interval=interval,
            limit=limit
        )
    """
    # Simulate data retrieval
    await asyncio.sleep(0.05)  # Simulate network latency
    
    return {
        "symbol": symbol,
        "interval": interval,
        "klines": [
            {
                "timestamp": "2026-04-01T09:30:00Z",
                "open": 185.50,
                "high": 186.20,
                "low": 185.10,
                "close": 186.00,
                "volume": 1250000
            }
        ]
    }


# ============================================================
# STARTUP
# ============================================================

if __name__ == "__main__":
    port = int(os.environ.get("PORT", 8000))
    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=port,
        log_level="info"
    )

---

Summary

The journey from print() to structlog is not cosmetic. It is a fundamental shift in how we treat log data: from human-readable strings to machine-queriable events.

For financial APIs specifically, structured logging delivers measurable benefits:

1. Incident response time: A correlation ID query in Kibana traces an order across all services in under a second.
2. Audit compliance: JSON logs with field-level typing satisfy regulatory requirements for immutable, timestamped records.
3. Performance monitoring: Latency fields enable percentile calculations without post-processing.
4. Security: Redaction processors prevent PII from entering log streams.
5. Scalability: JSON serialization is CPU-efficient at high throughput; sampling prevents pipeline overload during volatile markets.

The configuration shown in this article is production-ready for a financial data API. It includes correlation ID propagation, sensitive data redaction, latency monitoring, and ELK integration. Adapt the processors to your specific compliance requirements and deployment constraints.

Logging is infrastructure. Treat it with the same rigor you apply to your trading logic.

---

Next Steps

If you're building a trading system and need reliable market data alongside observability tooling:

1. Sign up at tickdb.ai for API access (free tier available)
2. Explore the documentation for WebSocket depth channels and historical kline endpoints
3. Clone the structured logging example from this article and adapt the redaction rules to your compliance requirements

If you're instrumenting an existing Python service:

1. Install structlog: pip install structlog
2. Copy the processor chain from the production configuration section
3. Replace string interpolation with structured key-value pairs
4. Deploy Filebeat with the provided configuration to ship JSON logs to your ELK cluster

If you need historical data for backtesting your logging-aware strategies:

Reach out to enterprise@tickdb.ai for institutional data plans covering 10+ years of cleaned US equity OHLCV data.

---

This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results. Structured logging is an engineering practice applicable to any software system, regardless of the domain.