AlphaEngine/position_sizer.py at master · malaythakur/AlphaEngine · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
"""Position sizing based on volatility (ATR)."""

from __future__ import annotations

from typing import Optional

from loguru import logger

from config import Config, settings


class PositionSizer:
    """Calculate position size based on volatility and risk parameters."""

    def __init__(self, config: Optional[Config] = None) -> None:
        self._cfg = config or settings

    def calculate_size(self, atr: Optional[float], entry_price: float) -> float:
        """Calculate position size based on ATR.

        Smaller ATR (calm market) → larger position
        Larger ATR (volatile market) → smaller position

        Args:
            atr: Average True Range in USD (None = use base size)
            entry_price: Entry price for the trade

        Returns:
            Position size in BTC
        """
        if not self._cfg.USE_VOLATILITY_SIZING or atr is None or atr <= 0:
            return float(self._cfg.TRADE_QTY)

        # Scale inversely to ATR
        # If ATR = BASE_ATR, use TRADE_QTY
        # If ATR < BASE_ATR (calm), increase size
        # If ATR > BASE_ATR (volatile), decrease size
        size = self._cfg.TRADE_QTY * (self._cfg.BASE_ATR_FOR_SIZING / atr)

        # Clamp to min/max
        size = max(self._cfg.MIN_POSITION_SIZE, min(self._cfg.MAX_POSITION_SIZE, size))

        logger.debug(
            "PositionSizer | atr={:.2f} | base_atr={:.2f} | size={:.6f} BTC | "
            "entry_price={:.2f}",
            atr,
            self._cfg.BASE_ATR_FOR_SIZING,
            size,
            entry_price,
        )

        return size

    def calculate_risk_adjusted_size(
        self,
        atr: Optional[float],
        entry_price: float,
        stop_loss_price: float,
        account_equity: float,
        risk_pct: float = 1.0,
    ) -> float:
        """Calculate position size based on risk (Kelly-like approach).

        Risk = (entry - SL) * size = risk_pct% of account

        Args:
            atr: Average True Range
            entry_price: Entry price
            stop_loss_price: Stop loss price
            account_equity: Current account equity
            risk_pct: Risk per trade as % of account (default 1%)

        Returns:
            Position size in BTC
        """
        if entry_price <= stop_loss_price:
            logger.warning("PositionSizer | invalid SL: entry={} sl={}", entry_price, stop_loss_price)
            return self.calculate_size(atr, entry_price)

        # Risk in USDT
        risk_usdt = account_equity * (risk_pct / 100.0)

        # Risk per BTC
        risk_per_btc = entry_price - stop_loss_price

        # Size = risk_usdt / risk_per_btc
        size = risk_usdt / risk_per_btc

        # Apply volatility scaling
        if self._cfg.USE_VOLATILITY_SIZING and atr is not None and atr > 0:
            size = size * (self._cfg.BASE_ATR_FOR_SIZING / atr)

        # Clamp to min/max
        size = max(self._cfg.MIN_POSITION_SIZE, min(self._cfg.MAX_POSITION_SIZE, size))

        logger.debug(
            "PositionSizer (risk-adjusted) | atr={:.2f} | risk_usdt={:.2f} | "
            "risk_per_btc={:.2f} | size={:.6f} BTC",
            atr or 0,
            risk_usdt,
            risk_per_btc,
            size,
        )

        return size