Position Sizing Algorithms for AI Trading Agents

import numpy as np
from dataclasses import dataclass
from typing import Optional, List, Dict
from collections import deque
import logging

logger = logging.getLogger('PositionSizer')


@dataclass
class TradeRecord:
    """Single completed trade result."""
    pnl: float
    equity_before: float
    symbol: str
    side: str  # 'long' or 'short'


@dataclass
class OpenPosition:
    """Currently open position for heat tracking."""
    symbol: str
    side: str
    entry_price: float
    stop_price: float
    units: float
    dollar_risk: float  # always positive


class PositionSizer:
    """
    Calculates position sizes using Kelly, fixed fractional,
    and volatility-adjusted (ATR) methods.

    Tracks portfolio heat, enforces concentration limits,
    and adapts Kelly fraction based on rolling win-rate.
    """

    def __init__(
        self,
        equity: float,
        max_risk_pct: float = 0.02,
        max_portfolio_heat: float = 0.20,
        kelly_fraction: float = 0.25,
        max_concentration: float = 0.35,
        trade_history_len: int = 100,
    ):
        self.equity = equity
        self.max_risk_pct = max_risk_pct
        self.max_portfolio_heat = max_portfolio_heat
        self.kelly_fraction = kelly_fraction
        self.max_concentration = max_concentration
        self.trade_history: deque = deque(maxlen=trade_history_len)
        self.open_positions: Dict[str, OpenPosition] = {}

    # ── Core sizing methods ─────────────────────────────────────────── #

    def kelly(
        self,
        win_rate: Optional[float] = None,
        avg_win: Optional[float] = None,
        avg_loss: Optional[float] = None,
    ) -> float:
        """
        Compute fractional Kelly position size as fraction of equity.
        Derives stats from trade_history if not provided.
        Returns 0.0 if insufficient data or no edge detected.
        """
        if win_rate is None or avg_win is None or avg_loss is None:
            stats = self._compute_trade_stats()
            if stats is None:
                logger.warning('Insufficient trade history for Kelly')
                return 0.0
            win_rate, avg_win, avg_loss = stats

        if avg_loss == 0:
            return 0.0

        rr = avg_win / avg_loss
        q = 1.0 - win_rate
        full_kelly = win_rate - (q / rr)

        if full_kelly <= 0:
            logger.info(f'No edge detected (Kelly={full_kelly:.4f})')
            return 0.0

        fractional = full_kelly * self.kelly_fraction
        fractional = min(fractional, self.max_risk_pct * 5)

        logger.info(
            f'Kelly: WR={win_rate:.2%} RR={rr:.2f} '
            f'Full={full_kelly:.4f} Frac={fractional:.4f}'
        )
        return fractional

    def fixed_fractional(
        self,
        entry_price: float,
        stop_price: float,
        custom_risk_pct: Optional[float] = None,
    ) -> Dict:
        """
        Size position so stop hit = risk_pct loss of equity.
        Returns dict with units, dollar_risk, heat_after.
        Returns units=0 if position would breach heat limits.
        """
        risk_pct = custom_risk_pct or self.max_risk_pct
        dollar_risk = self.equity * risk_pct
        stop_dist = abs(entry_price - stop_price)

        if stop_dist == 0:
            return {'units': 0, 'dollar_risk': 0,
                    'reason': 'zero stop distance'}

        units = dollar_risk / stop_dist
        heat_added = dollar_risk / self.equity
        current_heat = self.portfolio_heat()

        if current_heat + heat_added > self.max_portfolio_heat:
            remaining = self.max_portfolio_heat - current_heat
            if remaining <= 0:
                logger.warning('Portfolio at max heat; rejecting position')
                return {'units': 0, 'dollar_risk': 0,
                        'reason': 'max heat reached'}
            dollar_risk = self.equity * remaining
            units = dollar_risk / stop_dist
            heat_added = remaining

        return {
            'units': units,
            'dollar_risk': dollar_risk,
            'heat_pct': heat_added,
            'heat_after': current_heat + heat_added,
            'position_value': units * entry_price,
        }

    def vol_adjusted(
        self,
        entry_price: float,
        atr: float,
        atr_multiplier: float = 2.0,
        custom_risk_pct: Optional[float] = None,
    ) -> Dict:
        """ATR-based volatility-adjusted position sizing."""
        stop_price = entry_price - (atr * atr_multiplier)
        return self.fixed_fractional(
            entry_price=entry_price,
            stop_price=stop_price,
            custom_risk_pct=custom_risk_pct,
        )

    # ── ATR helper ──────────────────────────────────────────────────── #

    @staticmethod
    def compute_atr(
        highs: List[float],
        lows: List[float],
        closes: List[float],
        period: int = 14,
    ) -> float:
        """Compute ATR using Wilder's smoothing (EMA alpha=1/period)."""
        if len(closes) < period + 1:
            raise ValueError(f'Need at least {period+1} candles')
        true_ranges = []
        for i in range(1, len(closes)):
            tr = max(
                highs[i] - lows[i],
                abs(highs[i] - closes[i - 1]),
                abs(lows[i] - closes[i - 1]),
            )
            true_ranges.append(tr)
        atr = sum(true_ranges[:period]) / period
        alpha = 1.0 / period
        for tr in true_ranges[period:]:
            atr = alpha * tr + (1 - alpha) * atr
        return atr

    # ── Portfolio heat tracking ──────────────────────────────────────── #

    def portfolio_heat(self) -> float:
        total = sum(p.dollar_risk for p in self.open_positions.values())
        return total / self.equity if self.equity > 0 else 0.0

    def add_position(self, pos: OpenPosition) -> None:
        self.open_positions[pos.symbol] = pos

    def close_position(self, symbol: str, pnl: float) -> None:
        if symbol in self.open_positions:
            pos = self.open_positions.pop(symbol)
            self.trade_history.append(TradeRecord(
                pnl=pnl,
                equity_before=self.equity,
                symbol=symbol,
                side=pos.side,
            ))
            self.equity += pnl

    # ── Statistics helpers ───────────────────────────────────────────── #

    def _compute_trade_stats(self):
        """Returns (win_rate, avg_win, avg_loss) or None if <10 trades."""
        if len(self.trade_history) < 10:
            return None
        wins   = [t.pnl for t in self.trade_history if t.pnl > 0]
        losses = [abs(t.pnl) for t in self.trade_history if t.pnl < 0]
        if not wins or not losses:
            return None
        return (
            len(wins) / len(self.trade_history),
            sum(wins) / len(wins),
            sum(losses) / len(losses),
        )

    def rolling_win_rate(self, window: int = 20) -> float:
        recent = list(self.trade_history)[-window:]
        if not recent:
            return 0.5
        return sum(1 for t in recent if t.pnl > 0) / len(recent)

    def summary(self) -> Dict:
        stats = self._compute_trade_stats()
        return {
            'equity': self.equity,
            'portfolio_heat': self.portfolio_heat(),
            'open_positions': len(self.open_positions),
            'total_trades': len(self.trade_history),
            'rolling_win_rate': self.rolling_win_rate(),
            'kelly_recommended': self.kelly() if stats else None,
        }

import httpx
import asyncio
from position_sizer import PositionSizer, OpenPosition

BASE_URL = 'https://api.purpleflea.com'
API_KEY  = 'pf_live_your_key_here'

HEADERS = {
    'Authorization': f'Bearer {API_KEY}',
    'Content-Type': 'application/json',
}


async def get_wallet_equity(client: httpx.AsyncClient) -> float:
    resp = await client.get(f'{BASE_URL}/v1/wallet/balance', headers=HEADERS)
    resp.raise_for_status()
    return float(resp.json()['balance']['usd'])


async def get_ohlcv(client: httpx.AsyncClient, symbol: str, limit: int = 30):
    resp = await client.get(
        f'{BASE_URL}/v1/market/ohlcv',
        params={'symbol': symbol, 'interval': '1h', 'limit': limit},
        headers=HEADERS,
    )
    resp.raise_for_status()
    return resp.json()['candles']


async def place_sized_order(
    client: httpx.AsyncClient,
    sizer: PositionSizer,
    symbol: str,
    side: str,
    entry_price: float,
    method: str = 'vol_adjusted',
) -> dict:
    """
    Compute position size and place order on Purple Flea.
    Supported methods: 'vol_adjusted', 'kelly', 'fixed'
    """
    sizer.equity = await get_wallet_equity(client)

    if method == 'vol_adjusted':
        candles = await get_ohlcv(client, symbol)
        highs  = [c['high'] for c in candles]
        lows   = [c['low'] for c in candles]
        closes = [c['close'] for c in candles]
        atr    = PositionSizer.compute_atr(highs, lows, closes)
        sizing = sizer.vol_adjusted(entry_price=entry_price, atr=atr)

    elif method == 'kelly':
        kelly_f = sizer.kelly()
        dollar_pos = sizer.equity * kelly_f
        stop_pct = 0.05  # 5% stop for Kelly positions
        stop_price = entry_price * (1 - stop_pct if side == 'buy' else 1 + stop_pct)
        sizing = {
            'units': dollar_pos / entry_price,
            'dollar_risk': dollar_pos * stop_pct,
            'stop_price': stop_price,
        }

    else:  # fixed fractional
        stop_pct = 0.03
        stop_price = entry_price * (1 - stop_pct if side == 'buy' else 1 + stop_pct)
        sizing = sizer.fixed_fractional(entry_price, stop_price)

    if sizing.get('units', 0) == 0:
        return {'status': 'skipped', 'reason': sizing.get('reason')}

    resp = await client.post(
        f'{BASE_URL}/v1/orders',
        json={
            'symbol': symbol,
            'side': side,
            'quantity': round(sizing['units'], 6),
            'type': 'market',
            'stop_loss': sizing.get('stop_price'),
        },
        headers=HEADERS,
    )
    resp.raise_for_status()

    stop_p = sizing.get('stop_price', entry_price * 0.97)
    sizer.add_position(OpenPosition(
        symbol=symbol,
        side='long' if side == 'buy' else 'short',
        entry_price=entry_price,
        stop_price=stop_p,
        units=sizing['units'],
        dollar_risk=sizing['dollar_risk'],
    ))
    return resp.json()


async def main():
    async with httpx.AsyncClient() as client:
        equity = await get_wallet_equity(client)
        sizer  = PositionSizer(equity=equity, max_risk_pct=0.02)

        result = await place_sized_order(
            client, sizer,
            symbol='BTC-USD',
            side='buy',
            entry_price=85_000.0,
            method='vol_adjusted',
        )
        print(f'Order: {result}')
        print(f'Portfolio heat: {sizer.portfolio_heat():.2%}')
        print(f'Summary: {sizer.summary()}')

if __name__ == '__main__':
    asyncio.run(main())

class AdaptivePositionSizer(PositionSizer):
    """
    Extends PositionSizer with:
    - Dynamic risk % based on rolling win rate
    - Drawdown-triggered risk reduction
    - Performance-based Kelly multiplier scaling
    """

    def __init__(self, equity, peak_equity=None, **kwargs):
        super().__init__(equity, **kwargs)
        self.peak_equity = peak_equity or equity
        self.base_risk_pct = kwargs.get('max_risk_pct', 0.02)

    def adaptive_risk_pct(self) -> float:
        """
        Compute risk % dynamically from rolling win rate and drawdown.
        Floor: 0.25x base. Ceiling: 2x base.
        """
        wr = self.rolling_win_rate(window=20)
        dd = (1.0 - self.equity / self.peak_equity) if self.peak_equity > 0 else 0

        # Win rate scale: linear from 0.5x at WR=40% to 1.5x at WR=70%
        wr_mult = max(0.5, min(1.5, 1.0 + (wr - 0.55) * 3.33))

        # Drawdown scale: ratchet down risk when underwater
        dd_mult = (
            0.25 if dd > 0.20 else
            0.50 if dd > 0.10 else
            0.75 if dd > 0.05 else
            1.00
        )

        adjusted = self.base_risk_pct * wr_mult * dd_mult
        return max(
            self.base_risk_pct * 0.25,
            min(self.base_risk_pct * 2.0, adjusted)
        )

    def update_peak(self):
        if self.equity > self.peak_equity:
            self.peak_equity = self.equity

    def fixed_fractional(self, entry_price, stop_price, custom_risk_pct=None):
        risk = custom_risk_pct or self.adaptive_risk_pct()
        return super().fixed_fractional(entry_price, stop_price, risk)


async def sync_trade_history(
    client: httpx.AsyncClient,
    sizer: AdaptivePositionSizer,
    limit: int = 50,
) -> None:
    """Load recent closed trades from wallet API into sizer."""
    resp = await client.get(
        f'{BASE_URL}/v1/wallet/transactions',
        params={'type': 'trade_pnl', 'limit': limit},
        headers=HEADERS,
    )
    resp.raise_for_status()
    for tx in resp.json()['transactions']:
        sizer.trade_history.append(TradeRecord(
            pnl=float(tx['amount']),
            equity_before=float(tx['balance_before']),
            symbol=tx.get('symbol', 'UNKNOWN'),
            side=tx.get('side', 'long'),
        ))
    sizer.equity = await get_wallet_equity(client)
    sizer.update_peak()
    print(f'WR={sizer.rolling_win_rate():.2%} | Adaptive risk={sizer.adaptive_risk_pct():.3%}')

class CircuitBreaker:
    """Hard stops that override all position sizer decisions."""

    def __init__(
        self,
        daily_loss_limit: float = 0.05,
        session_loss_limit: float = 0.10,
        consecutive_loss_limit: int = 6,
        max_drawdown_halt: float = 0.20,
    ):
        self.daily_loss_limit = daily_loss_limit
        self.session_loss_limit = session_loss_limit
        self.consecutive_loss_limit = consecutive_loss_limit
        self.max_drawdown_halt = max_drawdown_halt
        self.session_start = None
        self.day_start = None
        self.consecutive_losses = 0
        self.halted = False
        self.halt_reason = None

    def check(self, equity: float, peak: float) -> bool:
        """Returns True if trading allowed, False if halted."""
        if self.halted:
            return False
        if self.session_start is None:
            self.session_start = equity
            self.day_start = equity
            return True

        session_loss = (self.session_start - equity) / self.session_start
        day_loss     = (self.day_start - equity) / self.day_start
        drawdown     = (peak - equity) / peak if peak > 0 else 0

        if day_loss >= self.daily_loss_limit:
            return self._halt(f'Daily loss limit: {day_loss:.2%}')
        if session_loss >= self.session_loss_limit:
            return self._halt(f'Session loss limit: {session_loss:.2%}')
        if drawdown >= self.max_drawdown_halt:
            return self._halt(f'Max drawdown: {drawdown:.2%}')
        if self.consecutive_losses >= self.consecutive_loss_limit:
            return self._halt(f'{self.consecutive_losses} consecutive losses')
        return True

    def record_trade(self, won: bool) -> None:
        self.consecutive_losses = 0 if won else self.consecutive_losses + 1

    def _halt(self, reason: str) -> bool:
        self.halted = True
        self.halt_reason = reason
        logger.critical(f'CIRCUIT BREAKER: {reason}')
        return False