# State Management Patterns

> State Management Patterns

- Skill: `khanh-vu/state-management-patterns` (Agent Skill)
- Install (CLI): `npx skillmds@latest add khanh-vu/state-management-patterns`
- Raw SKILL.md: https://api.skillmd.com/api/skills/khanh-vu/state-management-patterns/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: khanh-vu (https://skillmd.com/u/khanh-vu)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/khanh-vu/state-management-patterns

---

# State Management Patterns

Robust state management for distributed cryptocurrency trading systems.

## Position Reconciliation

```python
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
```

## Idempotency Pattern for Orders

```python
import hashlib
import uuid

class IdempotentOrderManager:
    """
    Ensure orders are only placed once, even with retries or duplicate requests
    Uses idempotency keys to prevent duplicate order submission
    """

    def __init__(self, exchange_connector, cache):
        self.exchange = exchange_connector
        self.cache = cache
        self.idempotency_ttl = 86400  # 24 hours

    async def place_order_idempotent(
        self,
        symbol: str,
        side: str,
        order_type: str,
        amount: Decimal,
        price: Optional[Decimal] = None,
        idempotency_key: Optional[str] = None
    ) -> Dict:
        """
        Place order with idempotency guarantee

        If called multiple times with same idempotency_key,
        returns the original order without placing duplicate
        """
        # Generate idempotency key if not provided
        if not idempotency_key:
            idempotency_key = self._generate_idempotency_key(
                symbol, side, order_type, amount, price
            )

        # Check if order already placed with this key
        cached_order = await self.cache.get(f"order:idempotency:{idempotency_key}")

        if cached_order:
            logger.info(
                f"Idempotent order already placed: {idempotency_key} -> {cached_order['id']}"
            )
            return cached_order

        # Place order
        try:
            if order_type == 'market':
                order = await self.exchange.create_market_order(
                    symbol=symbol,
                    side=side,
                    amount=float(amount)
                )
            else:
                order = await self.exchange.create_limit_order(
                    symbol=symbol,
                    side=side,
                    amount=float(amount),
                    price=float(price)
                )

            # Cache order with idempotency key
            await self.cache.set(
                f"order:idempotency:{idempotency_key}",
                order,
                ttl=self.idempotency_ttl
            )

            logger.info(
                f"Order placed with idempotency key {idempotency_key}: {order['id']}"
            )

            return order

        except Exception as e:
            logger.error(f"Order placement failed: {e}")
            raise

    def _generate_idempotency_key(
        self,
        symbol: str,
        side: str,
        order_type: str,
        amount: Decimal,
        price: Optional[Decimal],
        strategy_id: str = "default",
        signal_id: str = None
    ) -> str:
        """
        Generate deterministic idempotency key from order parameters

        CRITICAL: Does NOT include timestamp - must be deterministic across retries
        Uses strategy_id and signal_id for uniqueness instead of time
        """
        # Use signal_id if provided, otherwise generate from parameters
        if signal_id is None:
            # For manual orders, use parameter-based ID
            signal_id = f"{symbol}_{side}_{amount}_{price}"

        key_data = f"{strategy_id}:{signal_id}:{symbol}:{side}:{order_type}:{amount}:{price}"
        key_hash = hashlib.sha256(key_data.encode()).hexdigest()[:16]

        return key_hash
```

## Event Sourcing for Trade History

```python
from enum import Enum
from typing import List
import json

class EventType(Enum):
    ORDER_PLACED = "order_placed"
    ORDER_FILLED = "order_filled"
    ORDER_PARTIALLY_FILLED = "order_partially_filled"
    ORDER_CANCELLED = "order_cancelled"
    POSITION_OPENED = "position_opened"
    POSITION_CLOSED = "position_closed"
    POSITION_MODIFIED = "position_modified"

@dataclass
class Event:
    event_id: str
    event_type: EventType
    timestamp: datetime
    aggregate_id: str  # Order ID or Position ID
    data: Dict
    version: int

class EventStore:
    """
    Event sourcing for complete audit trail
    Reconstruct state from event stream
    """

    def __init__(self, database):
        self.db = database

    async def append_event(
        self,
        event_type: EventType,
        aggregate_id: str,
        data: Dict
    ) -> Event:
        """Append event to event store"""
        # Get current version for aggregate
        current_version = await self._get_aggregate_version(aggregate_id)

        event = Event(
            event_id=str(uuid.uuid4()),
            event_type=event_type,
            timestamp=datetime.utcnow(),
            aggregate_id=aggregate_id,
            data=data,
            version=current_version + 1
        )

        # Store event
        await self.db.insert_event({
            'event_id': event.event_id,
            'event_type': event.event_type.value,
            'timestamp': event.timestamp,
            'aggregate_id': event.aggregate_id,
            'data': json.dumps(data),
            'version': event.version
        })

        logger.debug(f"Event appended: {event.event_type.value} for {aggregate_id}")

        return event

    async def get_events(
        self,
        aggregate_id: str,
        since_version: int = 0
    ) -> List[Event]:
        """Retrieve events for aggregate"""
        rows = await self.db.query(
            """
            SELECT event_id, event_type, timestamp, aggregate_id, data, version
            FROM events
            WHERE aggregate_id = $1 AND version > $2
            ORDER BY version ASC
            """,
            aggregate_id,
            since_version
        )

        events = []
        for row in rows:
            events.append(Event(
                event_id=row['event_id'],
                event_type=EventType(row['event_type']),
                timestamp=row['timestamp'],
                aggregate_id=row['aggregate_id'],
                data=json.loads(row['data']),
                version=row['version']
            ))

        return events

    async def reconstruct_order_state(self, order_id: str) -> Dict:
        """Reconstruct current order state from event stream"""
        events = await self.get_events(order_id)

        state = {
            'order_id': order_id,
            'status': 'unknown',
            'filled': Decimal('0'),
            'remaining': Decimal('0'),
            'fills': []
        }

        for event in events:
            if event.event_type == EventType.ORDER_PLACED:
                state['status'] = 'open'
                state['remaining'] = Decimal(str(event.data['amount']))
                state['symbol'] = event.data['symbol']
                state['side'] = event.data['side']

            elif event.event_type == EventType.ORDER_FILLED:
                state['status'] = 'filled'
                state['filled'] = Decimal(str(event.data['filled_amount']))
                state['remaining'] = Decimal('0')
                state['avg_price'] = Decimal(str(event.data['avg_price']))

            elif event.event_type == EventType.ORDER_PARTIALLY_FILLED:
                fill_amount = Decimal(str(event.data['fill_amount']))
                state['filled'] += fill_amount
                state['remaining'] -= fill_amount
                state['fills'].append(event.data)

            elif event.event_type == EventType.ORDER_CANCELLED:
                state['status'] = 'cancelled'

        return state
```

## Optimistic Locking for Concurrent Updates

```python
class OptimisticLockError(Exception):
    """Raised when optimistic lock fails due to concurrent modification"""
    pass

class OptimisticLockManager:
    """
    Optimistic locking for safe concurrent updates
    Uses version numbers to detect conflicts
    """

    # Whitelist of allowed tables to prevent SQL injection
    ALLOWED_TABLES = {'orders', 'positions', 'balances', 'trades', 'strategies'}

    async def update_with_optimistic_lock(
        self,
        table: str,
        record_id: str,
        updates: Dict,
        current_version: int
    ):
        """
        Update record with optimistic locking

        Raises OptimisticLockError if version mismatch (concurrent update)
        Raises ValueError if table name is not whitelisted
        """
        # Validate table name to prevent SQL injection
        if table not in self.ALLOWED_TABLES:
            raise ValueError(
                f"Invalid table name: {table}. "
                f"Allowed tables: {', '.join(self.ALLOWED_TABLES)}"
            )

        # Attempt update with version check
        result = await self.db.execute(
            f"""
            UPDATE {table}
            SET
                {', '.join(f"{k} = ${i+2}" for i, k in enumerate(updates.keys()))},
                version = version + 1,
                updated_at = NOW()
            WHERE id = $1 AND version = ${len(updates) + 2}
            RETURNING version
            """,
            record_id,
            *updates.values(),
            current_version
        )

        if not result:
            # Version mismatch - concurrent update detected
            raise OptimisticLockError(
                f"Concurrent modification detected for {table}:{record_id}. "
                f"Expected version {current_version}, but it was modified."
            )

        new_version = result[0]['version']
        logger.debug(f"Optimistic lock success: {table}:{record_id} updated to v{new_version}")

        return new_version

    async def update_with_retry(
        self,
        table: str,
        record_id: str,
        update_func,
        max_retries: int = 3
    ):
        """
        Update with automatic retry on optimistic lock failure
        update_func receives current record and returns updates dict
        """
        for attempt in range(max_retries):
            try:
                # Fetch current record
                record = await self.db.get_record(table, record_id)

                # Generate updates
                updates = update_func(record)

                # Attempt update
                new_version = await self.update_with_optimistic_lock(
                    table,
                    record_id,
                    updates,
                    record['version']
                )

                return new_version

            except OptimisticLockError:
                if attempt == max_retries - 1:
                    raise
                logger.warning(
                    f"Optimistic lock retry {attempt + 1}/{max_retries} for {table}:{record_id}"
                )
                await asyncio.sleep(0.1 * (attempt + 1))  # Exponential backoff
```

---
**Critical patterns** for maintaining state consistency in distributed trading systems.