claude-force: Skill for Claude Code

State Management Patterns

Robust state management for distributed cryptocurrency trading systems.

Position Reconciliation

from dataclasses import dataclass
from decimal import Decimal
from typing import Dict, List, Optional
from datetime import datetime

@dataclass
class Position:
    symbol: str
    amount: Decimal
    entry_price: Decimal
    current_price: Decimal
    unrealized_pnl: Decimal
    realized_pnl: Decimal
    leverage: Decimal
    liquidation_price: Optional[Decimal]
    last_updated: datetime

class PositionReconciliation:
    """
    Reconcile positions between exchange, database, and in-memory state
    Critical for preventing duplicate orders and state drift
    """

    def __init__(self, exchange_connector, database, cache):
        self.exchange = exchange_connector
        self.db = database
        self.cache = cache

    async def reconcile_positions(self) -> Dict[str, Position]:
        """
        Full position reconciliation across all data sources
        Returns: Reconciled positions (source of truth)
        """
        logger.info("Starting position reconciliation...")

        # Fetch positions from all sources
        exchange_positions = await self._fetch_exchange_positions()
        db_positions = await self.db.get_positions()
        cache_positions = await self.cache.get_positions()

        # Build position map keyed by symbol
        reconciled = {}

        # Use exchange as source of truth
        for symbol, exchange_pos in exchange_positions.items():
            db_pos = db_positions.get(symbol)
            cache_pos = cache_positions.get(symbol)

            # Detect discrepancies
            discrepancies = []

            if db_pos and abs(db_pos.amount - exchange_pos.amount) > Decimal('0.0001'):
                discrepancies.append(
                    f"DB amount mismatch: {db_pos.amount} vs exchange {exchange_pos.amount}"
                )

            if cache_pos and abs(cache_pos.amount - exchange_pos.amount) > Decimal('0.0001'):
                discrepancies.append(
                    f"Cache amount mismatch: {cache_pos.amount} vs exchange {exchange_pos.amount}"
                )

            # Update DB and cache to match exchange
            if discrepancies:
                logger.warning(
                    f"Position discrepancy for {symbol}: {', '.join(discrepancies)}"
                )

                await self.db.update_position(exchange_pos)
                await self.cache.set_position(symbol, exchange_pos)

            reconciled[symbol] = exchange_pos

        # Check for positions in DB/cache but not on exchange (stale data)
        all_symbols = set(exchange_positions.keys()) | set(db_positions.keys()) | set(cache_positions.keys())

        for symbol in all_symbols:
            if symbol not in exchange_positions:
                if symbol in db_positions or symbol in cache_positions:
                    logger.warning(f"Found stale position for {symbol} - removing from DB/cache")
                    await self.db.delete_position(symbol)
                    await self.cache.delete_position(symbol)

        logger.info(f"Position reconciliation complete: {len(reconciled)} positions")

        return reconciled

    async def _fetch_exchange_positions(self) -> Dict[str, Position]:
        """Fetch positions from exchange and convert to Position objects"""
        raw_positions = await self.exchange.fetch_positions()

        positions = {}
        for pos in raw_positions:
            if pos['contracts'] == 0:
                continue  # Skip closed positions

            position = Position(
                symbol=pos['symbol'],
                amount=Decimal(str(pos['contracts'])),
                entry_price=Decimal(str(pos['entryPrice'])),
                current_price=Decimal(str(pos['markPrice'])),
                unrealized_pnl=Decimal(str(pos['unrealizedPnl'])),
                realized_pnl=Decimal(str(pos.get('realizedPnl', 0))),
                leverage=Decimal(str(pos.get('leverage', 1))),
                liquidation_price=Decimal(str(pos['liquidationPrice'])) if pos.get('liquidationPrice') else None,
                last_updated=datetime.utcnow()
            )

            positions[pos['symbol']] = position

        return positions