FiveTwenty SDK Best Practices

Essential patterns and practices for building robust applications with the FiveTwenty SDK.

Client Architecture Patterns

Context Manager Usage

Always use context managers for proper resource cleanup:

import os

from fivetwenty import AsyncClient, Client, Environment

# Step 1: Configure authentication and account parameters
# Environment variables provide secure token management
token = os.getenv("OANDA_TOKEN")  # API token from environment for security
account_id = "101-001-0000000-001"  # Replace with your OANDA account ID

# Step 2: AsyncClient usage with proper resource management
async def example_async():
    """Demonstrate AsyncClient usage with automatic resource cleanup."""
    # Context manager ensures proper connection cleanup and resource management
    async with AsyncClient(token=token, environment=Environment.PRACTICE) as client:
        # Step 3: HTTP connections and session cleanup handled automatically
        # AsyncClient manages connection pooling and graceful shutdown
        print(f"Config AsyncClient initialized with practice environment")
        print(f"Global HTTP session created with connection pooling")

        # Step 4: Execute API call with automatic error handling
        account = await client.accounts.get_account(account_id)
        print(f"Success Account retrieved: {account.id}")
        return account
        # Note: Client automatically cleaned up on exit from context manager

# Step 5: Sync Client usage for non-async applications
def example_sync():
    """Demonstrate sync Client usage with background thread management."""
    # Context manager handles background asyncio thread and bounded queues
    with Client(token=token, environment=Environment.PRACTICE) as client:
        # Step 6: Background thread and queue management handled automatically
        # Sync client runs AsyncClient in background thread with queue-based communication
        print(f"Config Sync Client initialized with background thread")
        print(f"Satellite Bounded queues prevent memory overflow in sync wrapper")

        # Step 7: Synchronous API call - internally queued to async thread
        account = client.accounts.get_account(account_id)
        print(f"Success Account retrieved synchronously: {account.id}")
        return account
        # Note: Background thread and queues automatically cleaned up on exit

Connection Reuse

Reuse client instances across multiple operations:

import os

from fivetwenty import AsyncClient, Environment

# Step 1: Configure authentication parameters
token = os.getenv("OANDA_TOKEN")  # API token from environment
account_id = "101-001-0000000-001"  # Replace with your OANDA account ID

# Step 2: Success Good: Reuse client for multiple operations (RECOMMENDED)
async def good_example():
    """Optimal pattern: Single client instance for multiple operations."""
    # Single context manager for all related operations
    async with AsyncClient(token=token, environment=Environment.PRACTICE) as client:
        print(f"Config Single client instance created for multiple operations")
        print(f"Global HTTP connection pool shared across all API calls")

        # Step 3: Execute multiple operations using same client and connections
        # Connection pooling provides significant performance benefits
        account = await client.accounts.get_account(account_id)     # Reuses connection
        positions = await client.positions.get_positions(account_id)  # Reuses connection
        orders = await client.orders.get_orders(account_id)        # Reuses connection

        print(f"Success Retrieved account, positions, and orders using shared connections")
        print(f"Lightning Connection reuse eliminates handshake overhead")
        return account, positions, orders
        # Single cleanup for all operations

# Step 4: Error Bad: Create new client for each operation (AVOID)
async def get_account():
    """Anti-pattern: New client instance per operation wastes resources."""
    # New context manager creates unnecessary overhead
    async with AsyncClient(token=token, environment=Environment.PRACTICE) as client:
        print(f"⚠️ New client created for single operation")
        print(f"💸 Connection overhead: handshake, auth, session setup")
        return await client.accounts.get_account(account_id)
        # Immediate cleanup - connections destroyed and recreated for each call
    # NOTE: This pattern multiplies network overhead and reduces performance

Concurrent Operations

Use asyncio.gather() for concurrent API calls:

import asyncio
import os

from fivetwenty import AsyncClient, Environment

# Step 1: Configure authentication parameters
token = os.getenv("OANDA_TOKEN")  # API token from environment
account_id = "101-001-0000000-001"  # Replace with your OANDA account ID

async def concurrent_example() -> None:
    """Demonstrate concurrent vs sequential API operations for optimal performance."""
    async with AsyncClient(token=token, environment=Environment.PRACTICE) as client:
        print(f"Starting Starting concurrent operations demonstration...")

        # Step 2: Success Efficient: Concurrent requests using asyncio.gather
        # Multiple API calls execute simultaneously, reducing total latency
        start_time = asyncio.get_event_loop().time()
        account, positions, orders = await asyncio.gather(
            client.accounts.get_account(account_id),     # Executes concurrently
            client.positions.get_positions(account_id),  # Executes concurrently
            client.orders.get_orders(account_id)         # Executes concurrently
        )
        concurrent_time = asyncio.get_event_loop().time() - start_time
        print(f"Lightning Concurrent execution: {concurrent_time:.3f} seconds")
        print(f"Global Three API calls executed simultaneously")

        # Step 3: Error Inefficient: Sequential requests (for comparison)
        # Each API call waits for previous to complete - multiplies latency
        start_time = asyncio.get_event_loop().time()
        account = await client.accounts.get_account(account_id)      # Waits for completion
        positions = await client.positions.get_positions(account_id) # Waits for completion
        orders = await client.orders.get_orders(account_id)         # Waits for completion
        sequential_time = asyncio.get_event_loop().time() - start_time
        print(f"🐌 Sequential execution: {sequential_time:.3f} seconds")
        print(f"Wait Each API call blocks the next")

        # Step 4: Display performance comparison
        performance_gain = (sequential_time - concurrent_time) / sequential_time * 100
        print(f"\nData Performance Analysis:")
        print(f"   Concurrent: {concurrent_time:.3f}s")
        print(f"   Sequential: {sequential_time:.3f}s")
        print(f"   Improvement: {performance_gain:.1f}% faster with concurrency")
        print(f"Note Use asyncio.gather() for independent API calls")

        return account, positions, orders

Financial Precision

Decimal Usage

Always use Decimal for financial calculations:

from decimal import Decimal

# Example account balance (typically from OANDA API)
account_balance = Decimal("10000.00")

# Good: Exact precision
position_size = Decimal(10000)
risk_amount = account_balance * Decimal("0.02")  # 2% risk
stop_distance = Decimal("0.0050")  # 50 pips

# Bad: Floating point errors (commented out)
# position_size = 10000.0
# risk_amount = account_balance * Decimal("0.02")
# stop_distance = 0.0050

Price Calculation Precision

Handle OANDA's price precision requirements:

from decimal import Decimal

# OANDA price precision (5 decimal places for majors)
price = Decimal("1.08456").quantize(Decimal("0.00001"))

# Position sizing with exact arithmetic
def calculate_position_size(
    account_balance: str,  # AccountUnits from OANDA
    risk_percentage: Decimal,
    stop_loss_pips: int,
    pip_value: Decimal,
) -> Decimal:
    """Calculate position size with exact precision."""
    balance = Decimal(account_balance)
    risk_amount = balance * (risk_percentage / 100)
    risk_per_unit = stop_loss_pips * pip_value
    position_size = risk_amount / risk_per_unit
    return position_size.quantize(Decimal(1))  # Round to whole units

Field Type Handling

FiveTwenty automatically converts between Decimal and string fields:

import os
from decimal import Decimal
from fivetwenty import AsyncClient, Environment

# Setup
token = os.getenv("OANDA_TOKEN")
account_id = "101-001-0000000-001"

async def example_order() -> tuple[object, object]:
    async with AsyncClient(token=token, environment=Environment.PRACTICE) as client:
        # SDK handles conversion automatically
        order = await client.orders.post_limit_order(
            account_id=account_id,
            instrument="EUR_USD",
            units=Decimal(10000),  # Converted to string for API
            price=Decimal("1.0850")  # Converted to string for API
        )

        # Response fields are properly typed
        filled_price = Decimal(order.order_fill_transaction.price)  # string -> Decimal
        return order, filled_price

Error Handling Patterns

Exception Hierarchy

Use specific exception types for targeted handling:

import asyncio
import os
from fivetwenty import AsyncClient, Environment
from fivetwenty.exceptions import FiveTwentyError

# Step 1: Configure authentication parameters
token = os.getenv("OANDA_TOKEN")  # API token from environment
account_id = "101-001-0000000-001"  # Replace with your OANDA account ID

async def example_error_handling() -> None:
    """Demonstrate comprehensive error handling for robust trading applications."""
    async with AsyncClient(token=token, environment=Environment.PRACTICE) as client:
        try:
            # Step 2: Attempt order placement with potential for various errors
            print(f"Analysis Attempting market order placement...")
            order = await client.orders.post_market_order(
                account_id=account_id,
                instrument="EUR_USD",
                units=1000
            )
            print(f"Success Order placed successfully: {order.id}")

        # Step 3: Handle specific OANDA API errors with targeted responses
        except FiveTwentyError as e:
            error_message = str(e)
            print(f"⚠️ OANDA API Error: {error_message}")

            # Step 4: Handle insufficient margin errors
            if "INSUFFICIENT_MARGIN" in error_message:
                print(f"Balance Insufficient margin detected")
                print(f"Config Reducing position size and retrying...")
                await handle_margin_error(e)

            # Step 5: Handle invalid instrument errors
            elif "INVALID_INSTRUMENT" in error_message:
                print(f"⚠️ Invalid instrument: EUR_USD not available")
                print(f"⏭️ Skipping this instrument and continuing...")
                return None

            # Step 6: Handle rate limiting errors
            elif "RATE_LIMIT" in error_message or "TOO_MANY_REQUESTS" in error_message:
                print(f"Time Rate limit exceeded - respecting API limits")
                retry_after = getattr(e, 'retry_after', 60)  # Default 60 seconds
                print(f"Wait Waiting {retry_after} seconds before retry...")
                await asyncio.sleep(retry_after)

            # Step 7: Handle server errors with exponential backoff
            elif "INTERNAL_SERVER_ERROR" in error_message or "503" in error_message:
                print(f"Tools OANDA server error detected")
                print(f"Processing Implementing exponential backoff retry...")
                await retry_with_backoff()

            else:
                print(f"Error Unhandled OANDA error: {error_message}")
                raise  # Re-raise unhandled errors

        # Step 8: Handle network and connection errors
        except ConnectionError as e:
            print(f"Global Network connection error: {e}")
            print(f"Config Check internet connectivity and retry...")
            await asyncio.sleep(5)  # Brief pause before potential retry

        # Step 9: Handle authentication errors
        except PermissionError as e:
            print(f"Secure Authentication error: {e}")
            print(f"Note Check API token validity and permissions")
            raise  # Don't retry auth errors

        return order

async def handle_margin_error(error) -> None:
    """Handle insufficient margin errors with position size reduction."""
    print(f"Tools Implementing margin error recovery strategy...")
    print(f"📉 Strategy: Reduce position size by 50%")
    print(f"Data Strategy: Check account balance and available margin")
    # Implementation would include account balance checks and position adjustment

async def retry_with_backoff() -> None:
    """Implement exponential backoff retry for server errors."""
    print(f"Processing Implementing exponential backoff retry strategy...")
    print(f"Time Delay pattern: 1s, 2s, 4s, 8s, 16s...")
    print(f"🎲 Adding jitter to prevent thundering herd")
    # Implementation would include actual retry logic with increasing delays

Retry Patterns

Implement exponential backoff for retryable errors:

import asyncio
import secrets
from collections.abc import Callable
from typing import Any

from fivetwenty.exceptions import FiveTwentyError

async def retry_with_backoff(
    operation: Callable[[], Any],
    max_retries: int = 3,
    base_delay: float = 1.0
) -> Any:
    """Retry operation with exponential backoff."""
    for attempt in range(max_retries):
        try:
            return await operation()
        except FiveTwentyError as e:
            if not (e.is_server_error or e.is_rate_limited):
                raise
            if attempt == max_retries - 1:
                raise

            # Exponential backoff with jitter
            delay = base_delay * (2 ** attempt)
            # Use secrets module for cryptographically secure random
            jitter = secrets.randbelow(int(0.3 * delay * 1000)) / 1000
            await asyncio.sleep(delay + jitter)
    return None  # Explicit return for all code paths

Streaming Best Practices

Stream Resource Management

Properly handle streaming connections:

import asyncio
from collections.abc import AsyncIterator
from fivetwenty import AsyncClient
from fivetwenty.exceptions import StreamStall

async def robust_price_stream(
    client: AsyncClient,
    account_id: str,
    instruments: list[str]
) -> AsyncIterator:
    """Streaming with proper error handling."""
    max_retries = 5
    retry_count = 0

    while retry_count < max_retries:
        try:
            async for price in client.pricing.get_pricing_stream(
                account_id=account_id,
                instruments=instruments
            ):
                # Reset retry count on successful data
                retry_count = 0
                yield price

        except StreamStall:
            retry_count += 1
            if retry_count >= max_retries:
                raise
            await asyncio.sleep(2 ** retry_count)  # Exponential backoff

Backpressure Management

Handle fast-moving data appropriately:

import os
from fivetwenty import AsyncClient, Client, Environment

# Setup
token = os.getenv("OANDA_TOKEN")
account_id = "101-001-0000000-001"

async def async_stream_example() -> None:
    async with AsyncClient(token=token, environment=Environment.PRACTICE) as client:
        # AsyncClient: Direct processing (no buffering)
        async for price in client.pricing.get_pricing_stream(
            account_id=account_id,
            instruments=["EUR_USD"]
        ):
            # Process immediately - don't block the stream
            await process_price_async(price)

def sync_stream_example() -> None:
    with Client(token=token, environment=Environment.PRACTICE) as client:
        # Sync Client: Bounded queues prevent memory issues
        for price in client.pricing.get_pricing_stream(
            account_id=account_id,
            instruments=["EUR_USD"]
        ):
            # Queue automatically manages backpressure
            process_price_sync(price)

async def process_price_async(price: object) -> None:
    """Process price data asynchronously."""
    pass

def process_price_sync(price: object) -> None:
    """Process price data synchronously."""
    pass

Environment Management

Environment-Specific Configuration

Use different settings for practice vs live:

import os

from fivetwenty import AsyncClient, Environment

def get_token(is_live: bool) -> str:
    """Get appropriate token for environment."""
    if is_live:
        return os.environ["OANDA_LIVE_TOKEN"]
    return os.environ["OANDA_PRACTICE_TOKEN"]

def create_client(is_live: bool = False) -> AsyncClient:
    """Create client with environment-appropriate settings."""
    env = Environment.LIVE if is_live else Environment.PRACTICE
    timeout = 30.0 if is_live else 10.0  # Longer timeout for live

    return AsyncClient(
        token=get_token(is_live),
        environment=env,
        timeout=timeout
    )

Token Security

Never hardcode tokens or log sensitive data:

import os
import logging

logger = logging.getLogger(__name__)

# Good: Environment variables
token = os.environ["OANDA_TOKEN"]

# Good: Masked logging
logger.info(f"Using token: {token[:8]}...")

# Bad: Hardcoded token
token = "your-actual-token-here"

# Bad: Token in logs
logger.info(f"Token: {token}")

Performance Optimization

Instrument Selection

Only stream instruments you actively use:

# Good: Specific instruments only
instruments = ["EUR_USD", "GBP_USD"]  # Only what you need

# Bad: Too many instruments - undefined variables example
# This would cause undefined variable errors:
# bases = ["EUR", "GBP", "USD", "JPY", "CHF", "CAD", "AUD", "NZD"]
# quotes = ["USD", "EUR", "GBP", "JPY", "CHF", "CAD", "AUD", "NZD"]
# instruments = [f"{base}_{quote}" for base in bases for quote in quotes]

Request Batching

Batch related operations when possible:

import os
from fivetwenty import AsyncClient, Environment

# Setup
token = os.getenv("OANDA_TOKEN")
account_id = "101-001-0000000-001"

async def batch_pricing_example() -> tuple[object, object, object, object]:
    async with AsyncClient(token=token, environment=Environment.PRACTICE) as client:
        # Good: Single request for multiple instruments
        prices = await client.pricing.get_pricing(
            account_id=account_id,
            instruments=["EUR_USD", "GBP_USD", "USD_JPY"]
        )

        # Less efficient: Multiple requests
        eur_usd = await client.pricing.get_pricing(account_id, ["EUR_USD"])
        gbp_usd = await client.pricing.get_pricing(account_id, ["GBP_USD"])
        usd_jpy = await client.pricing.get_pricing(account_id, ["USD_JPY"])

        return prices, eur_usd, gbp_usd, usd_jpy

Common Anti-Patterns

Avoid These Patterns

import os
import time
from fivetwenty import AsyncClient, Environment

# Setup
token = os.getenv("OANDA_TOKEN")
account_id = "101-001-0000000-001"

# Error Creating clients in loops
async def bad_pattern_example():
    symbols = ["EUR_USD", "GBP_USD"]
    for symbol in symbols:
        async with AsyncClient(token=token, environment=Environment.PRACTICE) as client:  # Expensive!
            _price = await client.pricing.get_pricing(account_id, [symbol])

# Error Blocking operations in async context
async def bad_async():
    time.sleep(1)  # Blocks entire event loop

# Error Float arithmetic for money
profit_loss = 1234.56 + 0.1  # Precision errors!

# Error Ignoring error details
async def bad_error_handling():
    async with AsyncClient(token=token, environment=Environment.PRACTICE) as client:
        try:
            _order = await client.orders.post_market_order(
                account_id=account_id,
                instrument="EUR_USD",
                units=1000
            )
        except Exception:
            pass  # Lost important error information

# Error Not handling rate limits
async def bad_rate_limit_handling():
    async with AsyncClient(token=token, environment=Environment.PRACTICE) as client:
        while True:
            await client.accounts.get_account(account_id)  # Will hit rate limits

Correct Alternatives

import asyncio
import logging
import os
from decimal import Decimal
from fivetwenty import AsyncClient, Environment
from fivetwenty.exceptions import FiveTwentyError

# Step 1: Configure structured logging for production applications
logger = logging.getLogger(__name__)

# Step 2: Configure authentication parameters securely
token = os.getenv("OANDA_TOKEN")  # Secure token from environment
account_id = "101-001-0000000-001"  # Replace with your account ID

# Step 3: Success Reuse client across operations for optimal performance
async def good_pattern_example():
    """Demonstrate optimal client reuse pattern for multiple operations."""
    symbols = ["EUR_USD", "GBP_USD"]
    print(f"Success Best Practice: Single client for multiple operations")

    # Single client instance handles all operations efficiently
    async with AsyncClient(token=token, environment=Environment.PRACTICE) as client:
        print(f"Config Client created once for {len(symbols)} operations")
        for symbol in symbols:
            # Each operation reuses the same HTTP connections
            _price = await client.pricing.get_pricing(account_id, [symbol])
            print(f"Balance Retrieved pricing for {symbol} using shared connection")
        print(f"Lightning Connection reuse provides significant performance benefit")

# Step 4: Success Async operations in async context for non-blocking execution
async def good_async():
    """Demonstrate proper async sleep vs blocking sleep."""
    print(f"Success Best Practice: Non-blocking async operations")
    print(f"Wait Starting 1-second non-blocking delay...")
    await asyncio.sleep(1)  # Non-blocking - allows other coroutines to run
    print(f"Success Non-blocking delay completed - event loop remained responsive")

# Step 5: Success Decimal arithmetic for financial precision
print(f"Success Best Practice: Decimal arithmetic for financial calculations")
profit_loss = Decimal("1234.56") + Decimal("0.1")  # Exact precision: 1234.66
print(f"Balance Precise calculation: 1234.56 + 0.1 = {profit_loss}")
print(f"✨ No floating-point precision errors with Decimal")

# Step 6: Success Specific error handling for robust applications
async def good_error_handling():
    """Demonstrate specific error handling with proper logging."""
    print(f"Success Best Practice: Specific error handling with logging")
    async with AsyncClient(token=token, environment=Environment.PRACTICE) as client:
        try:
            print(f"Analysis Attempting order placement with error handling...")
            _order = await client.orders.post_market_order(
                account_id=account_id,
                instrument="EUR_USD",
                units=1000
            )
            print(f"Success Order placed successfully")
        except FiveTwentyError as e:
            # Structured logging with specific error details
            logger.error(f"Order placement failed: {e}")
            print(f"Notes Error logged with specific details for debugging")
            print(f"Config Error handling allows graceful recovery")

# Step 7: Success Respect rate limits for API compliance
async def good_rate_limit_handling():
    """Demonstrate proper rate limit handling and compliance."""
    print(f"Success Best Practice: Respectful rate limit handling")
    async with AsyncClient(token=token, environment=Environment.PRACTICE) as client:
        try:
            print(f"Global Making API request with rate limit awareness...")
            await client.accounts.get_account(account_id)
            print(f"Success API request completed successfully")
        except FiveTwentyError as e:
            if not e.is_rate_limited:
                raise
            # Honor API rate limits to maintain good standing
            retry_after = e.retry_after or 60  # Use server-specified delay or default
            print(f"Time Rate limit hit - waiting {retry_after} seconds")
            print(f"🤝 Respecting API limits maintains good standing with OANDA")
            await asyncio.sleep(retry_after)
            print(f"Success Rate limit delay completed - ready for next request")

Order Validation Framework

Pre-Order Validation System

Implement comprehensive validation to prevent costly trading errors:

from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from enum import Enum
from typing import Any

from fivetwenty import AsyncClient


class ValidationSeverity(Enum):
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"
    CRITICAL = "critical"


@dataclass
class ValidationResult:
    is_valid: bool
    severity: ValidationSeverity
    rule_name: str
    message: str
    details: dict[str, Any] | None = None


class OrderValidator(ABC):
    """Base class for order validation rules."""

    def __init__(self, name: str, severity: ValidationSeverity = ValidationSeverity.ERROR) -> None:
        self.name = name
        self.severity = severity
        self.enabled = True

    @abstractmethod
    async def validate(self, order_params: dict[str, Any], context: dict[str, Any]) -> ValidationResult:
        """Validate order parameters and return result."""
        pass


class OrderValidationFramework:
    def __init__(self, client: AsyncClient, account_id: str) -> None:
        self.client = client
        self.account_id = account_id
        self.validators: list[OrderValidator] = []
        self.validation_history: list[object] = []

    def add_validator(self, validator: OrderValidator) -> None:
        """Add a validator to the framework."""
        self.validators.append(validator)

    async def validate_order(
        self,
        order_params: dict[str, Any],
        strict_mode: bool = True,
    ) -> dict[str, Any]:
        """Validate order against all registered validators."""
        validation_session = {
            "timestamp": datetime.utcnow(),
            "order_params": order_params,
            "results": [],
            "passed": True,
            "errors": [],
            "warnings": [],
        }

        # Build context for validators
        context = await self._build_validation_context()

        # Run all validators
        for validator in self.validators:
            if not validator.enabled:
                continue

            try:
                result = await validator.validate(order_params, context)
                validation_session["results"].append(result)

                if not result.is_valid:
                    if result.severity in [ValidationSeverity.ERROR, ValidationSeverity.CRITICAL]:
                        validation_session["errors"].append(result)
                        validation_session["passed"] = False
                    elif result.severity == ValidationSeverity.WARNING:
                        validation_session["warnings"].append(result)
                        if strict_mode:
                            validation_session["passed"] = False

            except Exception as e:
                error_result = ValidationResult(
                    is_valid=False,
                    severity=ValidationSeverity.CRITICAL,
                    rule_name=validator.name,
                    message=f"Validator failed: {e}",
                    details={"exception": str(e)},
                )
                validation_session["results"].append(error_result)
                validation_session["errors"].append(error_result)
                validation_session["passed"] = False

        # Store validation history
        self.validation_history.append(validation_session)
        return validation_session

    async def _build_validation_context(self) -> dict[str, Any]:
        """Build context information for validators."""
        try:
            # Get account information
            account = await self.client.accounts.get_account(account_id=self.account_id)
            positions = await self.client.positions.get_positions(account_id=self.account_id)
            orders = await self.client.orders.get_orders(account_id=self.account_id)

            return {
                "account": account,
                "positions": positions.positions,
                "pending_orders": orders.orders,
                "current_time": datetime.utcnow(),
                "account_balance": Decimal(account.balance),
                "margin_available": Decimal(account.margin_available),
                "margin_used": Decimal(account.margin_used),
            }

        except Exception as e:
            return {"error": str(e)}

Risk-Based Validators

class MaxPositionSizeValidator(OrderValidator):
    """Validate order doesn't exceed maximum position size limits."""

    def __init__(self, max_units_per_instrument: int, max_total_exposure: Decimal) -> None:
        super().__init__("MaxPositionSize", ValidationSeverity.ERROR)
        self.max_units_per_instrument = max_units_per_instrument
        self.max_total_exposure = max_total_exposure

    async def validate(self, order_params: dict[str, Any], context: dict[str, Any]) -> ValidationResult:
        """Validate position size limits."""
        instrument = order_params.get("instrument")
        units = int(order_params.get("units", 0))

        # Check individual instrument limit
        current_position_size = 0
        for position in context.get("positions", []):
            if position.instrument == instrument:
                if position.long.units != "0":
                    current_position_size += int(position.long.units)
                if position.short.units != "0":
                    current_position_size += abs(int(position.short.units))

        new_position_size = current_position_size + abs(units)

        if new_position_size > self.max_units_per_instrument:
            return ValidationResult(
                is_valid=False,
                severity=self.severity,
                rule_name=self.name,
                message=f"Position size {new_position_size} exceeds limit {self.max_units_per_instrument}",
                details={
                    "current_size": current_position_size,
                    "order_size": abs(units),
                    "new_size": new_position_size,
                    "limit": self.max_units_per_instrument
                }
            )

        return ValidationResult(
            is_valid=True,
            severity=ValidationSeverity.INFO,
            rule_name=self.name,
            message="Position size validation passed"
        )


class RiskPerTradeValidator(OrderValidator):
    """Validate risk per trade doesn't exceed limits."""

    def __init__(self, max_risk_per_trade: Decimal) -> None:
        super().__init__("RiskPerTrade", ValidationSeverity.WARNING)
        self.max_risk_per_trade = max_risk_per_trade

    async def validate(self, order_params: dict[str, Any], context: dict[str, Any]) -> ValidationResult:
        """Validate risk per trade."""
        units = order_params.get("units", 0)
        entry_price = order_params.get("price")
        stop_price = order_params.get("stop_loss_price")

        if not entry_price or not stop_price:
            return ValidationResult(
                is_valid=True,
                severity=ValidationSeverity.INFO,
                rule_name=self.name,
                message="No stop loss specified - cannot validate risk"
            )

        # Calculate risk amount
        stop_distance = abs(Decimal(str(entry_price)) - Decimal(str(stop_price)))
        risk_amount = abs(units) * stop_distance
        account_balance = context.get("account_balance", Decimal("0"))
        risk_percentage = risk_amount / account_balance if account_balance > 0 else Decimal("1")

        if risk_percentage > self.max_risk_per_trade:
            return ValidationResult(
                is_valid=False,
                severity=self.severity,
                rule_name=self.name,
                message=f"Risk {risk_percentage:.2%} exceeds limit {self.max_risk_per_trade:.2%}",
                details={
                    "risk_amount": risk_amount,
                    "risk_percentage": risk_percentage,
                    "limit": self.max_risk_per_trade,
                    "account_balance": account_balance
                }
            )

        return ValidationResult(
            is_valid=True,
            severity=ValidationSeverity.INFO,
            rule_name=self.name,
            message=f"Risk validation passed: {risk_percentage:.2%}"
        )

Production Error Recovery

Implement comprehensive error recovery for production systems:

from datetime import datetime
from enum import Enum
from typing import Any, Callable, Optional
import traceback
import asyncio

from fivetwenty.exceptions import FiveTwentyError


class ErrorCategory(Enum):
    NETWORK = "network"
    AUTHENTICATION = "authentication"
    VALIDATION = "validation"
    MARKET_DATA = "market_data"
    ORDER_EXECUTION = "order_execution"
    INSUFFICIENT_FUNDS = "insufficient_funds"
    RATE_LIMITING = "rate_limiting"
    SYSTEM = "system"


class TradingErrorHandler:
    def __init__(self, client: AsyncClient, account_id: str) -> None:
        self.client = client
        self.account_id = account_id
        self.error_handlers = {}
        self.error_history = []
        self.circuit_breaker_states = {}

    def register_error_handler(
        self,
        error_category: ErrorCategory,
        handler: Callable,
        max_retries: int = 3,
        retry_delay: int = 1
    ) -> None:
        """Register error handler for specific error category."""
        self.error_handlers[error_category] = {
            "handler": handler,
            "max_retries": max_retries,
            "retry_delay": retry_delay
        }

    async def handle_error(
        self,
        error: Exception,
        operation_context: dict[str, Any],
        error_category: Optional[ErrorCategory] = None
    ) -> dict[str, Any]:
        """Handle error with appropriate recovery strategy."""
        if not error_category:
            error_category = self._categorize_error(error)

        # Record error
        error_record = {
            "timestamp": datetime.utcnow(),
            "error_type": type(error).__name__,
            "error_message": str(error),
            "error_category": error_category,
            "operation_context": operation_context,
            "stack_trace": traceback.format_exc(),
            "recovery_attempted": False,
            "recovery_successful": False
        }

        self.error_history.append(error_record)

        # Check circuit breaker
        if self._should_circuit_break(error_category):
            error_record["circuit_breaker_triggered"] = True
            return error_record

        # Attempt recovery
        if error_category in self.error_handlers:
            handler_config = self.error_handlers[error_category]

            for attempt in range(handler_config["max_retries"]):
                try:
                    error_record["recovery_attempted"] = True
                    recovery_result = await handler_config["handler"](
                        error, operation_context, attempt
                    )

                    if recovery_result.get("success", False):
                        error_record["recovery_successful"] = True
                        error_record["recovery_result"] = recovery_result
                        break

                    await asyncio.sleep(handler_config["retry_delay"] * (2 ** attempt))

                except Exception as recovery_error:
                    error_record["recovery_error"] = str(recovery_error)

        return error_record

    def _categorize_error(self, error: Exception) -> ErrorCategory:
        """Categorize error based on type and message."""
        error_message = str(error).lower()

        if "network" in error_message or "connection" in error_message:
            return ErrorCategory.NETWORK
        if "authentication" in error_message or "unauthorized" in error_message:
            return ErrorCategory.AUTHENTICATION
        if "insufficient" in error_message and "margin" in error_message:
            return ErrorCategory.INSUFFICIENT_FUNDS
        if "rate limit" in error_message or "too many requests" in error_message:
            return ErrorCategory.RATE_LIMITING
        if "validation" in error_message or "invalid" in error_message:
            return ErrorCategory.VALIDATION
        if isinstance(error, FiveTwentyError):
            return ErrorCategory.ORDER_EXECUTION

        return ErrorCategory.SYSTEM

    def _should_circuit_break(self, error_category: ErrorCategory) -> bool:
        """Determine if circuit breaker should trigger."""
        recent_errors = [
            err for err in self.error_history[-10:]  # Last 10 errors
            if err["error_category"] == error_category
            and (datetime.utcnow() - err["timestamp"]).seconds < 300  # Last 5 minutes
        ]
        return len(recent_errors) >= 5

Testing Considerations

Mock Testing

Use proper mocking for unit tests:

from unittest.mock import AsyncMock
import pytest

# Mock response object
mock_response = AsyncMock()

# Example trading function to test
async def trading_function(client, account_id: str, instrument: str, units: int):
    return await client.orders.post_market_order(
        account_id=account_id,
        instrument=instrument,
        units=units
    )

@pytest.mark.asyncio
async def test_trading_logic():
    # Mock the client
    mock_client = AsyncMock()
    mock_client.orders.post_market_order.return_value = mock_response

    # Test your logic
    result = await trading_function(mock_client, "101-001-0000000-001", "EUR_USD", 1000)

    # Verify calls
    mock_client.orders.post_market_order.assert_called_once_with(
        account_id="101-001-0000000-001",
        instrument="EUR_USD",
        units=1000
    )

Integration Testing

Test against practice environment:

import os
import pytest
from fivetwenty import AsyncClient, Environment

@pytest.mark.integration
async def test_live_api():
    """Test against OANDA practice environment."""
    async with AsyncClient(
        token=os.environ["OANDA_PRACTICE_TOKEN"],
        environment=Environment.PRACTICE
    ) as client:
        # Test real API calls
        accounts = await client.accounts.get_accounts()
        if len(accounts) == 0:
            raise ValueError("Expected at least one account")

Validation Testing

Test your validation rules thoroughly:

@pytest.mark.asyncio
async def test_position_size_validator():
    # Create validator
    validator = MaxPositionSizeValidator(
        max_units_per_instrument=100000,
        max_total_exposure=Decimal("500000")
    )

    # Mock context with existing position
    context = {
        "positions": [
            MockPosition(instrument="EUR_USD", long_units="50000", short_units="0")
        ]
    }

    # Test order that would exceed limit
    order_params = {"instrument": "EUR_USD", "units": 60000}
    result = await validator.validate(order_params, context)

    if result.is_valid:
        raise ValueError("Expected validation to fail")
    if result.severity != ValidationSeverity.ERROR:
        raise ValueError(f"Expected severity ERROR, got '{result.severity}'")

Summary

Key principles for FiveTwenty SDK usage:

1. Use context managers for automatic resource cleanup
2. Always use Decimal for financial calculations
3. Handle specific exceptions rather than generic Exception
4. Reuse client instances for better performance
5. Implement proper retry logic with exponential backoff
6. Secure token management - never hardcode or log tokens
7. Test thoroughly in practice environment before live trading
8. Implement comprehensive validation - prevent costly errors before they occur
9. Build robust error recovery - handle failures gracefully with circuit breakers
10. Monitor and categorize errors - track patterns for system improvement

Following these patterns ensures robust, maintainable, and secure trading applications with comprehensive risk management and validation frameworks.