diff --git a/backend/app/services/risk_guard.py b/backend/app/services/risk_guard.py new file mode 100644 index 0000000..947bf71 --- /dev/null +++ b/backend/app/services/risk_guard.py @@ -0,0 +1,350 @@ +"""Position-sizing and risk-guard module. + +Implements the Vecna Trading Risk-Management Policy and Fractional-Kelly +Position Sizing rules: + + Per-trade risk cap equity × risk_pct (default 1 %; hard max 3 %) + Portfolio-heat limit total open risk ≤ 6 % of equity at all times + Drawdown circuit- at −15 % peak-to-trough → halve sizes; + breaker at −25 % → flatten all and halt trading + Fractional Kelly quarter-Kelly default, half-Kelly hard cap, + never full Kelly; f* = W − (1−W)/R + Concentration cap single-name ≤ 20 % of equity + Correlation cap at most 2 open positions with ρ > 0.7 +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum + +# --------------------------------------------------------------------------- +# Policy constants +# --------------------------------------------------------------------------- + +RISK_PCT_DEFAULT: float = 0.01 # 1 % per trade +RISK_PCT_MIN: float = 0.01 # floor +RISK_PCT_MAX: float = 0.03 # 3 % ceiling + +PORTFOLIO_HEAT_LIMIT: float = 0.06 # 6 % total open risk + +DD_HALVE_THRESHOLD: float = -0.15 # −15 % drawdown → halve sizes +DD_HALT_THRESHOLD: float = -0.25 # −25 % drawdown → flatten + halt + +KELLY_DEFAULT_FRACTION: float = 0.25 # quarter-Kelly default +KELLY_MAX_FRACTION: float = 0.50 # half-Kelly hard cap + +CONCENTRATION_CAP: float = 0.20 # 20 % single-name equity cap +CORRELATION_THRESHOLD: float = 0.70 # ρ > 0.7 counts as correlated +CORRELATION_MAX_POSITIONS: int = 2 # max correlated positions allowed + + +# --------------------------------------------------------------------------- +# Domain types +# --------------------------------------------------------------------------- + + +class CircuitBreakerState(str, Enum): + """Drawdown-based trading state.""" + + NORMAL = "normal" + HALVED = "halved" # −15 % drawdown: effective risk budget cut in half + HALTED = "halted" # −25 % drawdown: trading stopped, flatten all + + +@dataclass +class Position: + """Snapshot of a single open position for risk calculations.""" + + instrument: str + entry_price: float + stop_price: float + shares: float # units / shares / contracts + equity_at_entry: float # account equity when the position was sized + + @property + def risk_dollars(self) -> float: + """Dollar risk = |entry − stop| × shares.""" + return abs(self.entry_price - self.stop_price) * self.shares + + +@dataclass +class Portfolio: + """Live portfolio state fed into every pre-trade check.""" + + equity: float # current account equity ($) + peak_equity: float # highest equity seen (for drawdown) + open_positions: list[Position] = field(default_factory=list) + # Pairwise correlation matrix keyed by frozenset of instrument names + correlations: dict[frozenset[str], float] = field(default_factory=dict) + + @property + def drawdown(self) -> float: + """Peak-to-trough drawdown as a negative fraction (0.0 if peak ≤ 0).""" + if self.peak_equity <= 0: + return 0.0 + return (self.equity - self.peak_equity) / self.peak_equity + + @property + def circuit_breaker_state(self) -> CircuitBreakerState: + dd = self.drawdown + if dd <= DD_HALT_THRESHOLD: + return CircuitBreakerState.HALTED + if dd <= DD_HALVE_THRESHOLD: + return CircuitBreakerState.HALVED + return CircuitBreakerState.NORMAL + + @property + def total_open_risk(self) -> float: + """Total dollar risk across all open positions.""" + return sum(p.risk_dollars for p in self.open_positions) + + @property + def total_open_risk_pct(self) -> float: + """Total open risk as a fraction of equity; 0.0 when equity ≤ 0.""" + if self.equity <= 0: + return 0.0 + return self.total_open_risk / self.equity + + def exposure_for(self, instrument: str) -> float: + """Notional dollar exposure (entry × shares) for the given instrument.""" + return sum( + p.entry_price * p.shares + for p in self.open_positions + if p.instrument == instrument + ) + + def correlated_position_count(self, instrument: str) -> int: + """Count open positions whose correlation with *instrument* exceeds the threshold.""" + count = 0 + for pos in self.open_positions: + if pos.instrument == instrument: + continue + pair: frozenset[str] = frozenset({instrument, pos.instrument}) + if self.correlations.get(pair, 0.0) > CORRELATION_THRESHOLD: + count += 1 + return count + + +@dataclass(frozen=True) +class SizeResult: + """Output from :func:`compute_position_size`.""" + + shares: float # final position size in units + risk_dollars: float # dollar risk for this trade + kelly_fraction_used: float # fractional-Kelly value applied (0.0 if not used) + binding_rule: str # "kelly" | "risk_cap" | "heat_limit" | "concentration" + cb_state: CircuitBreakerState + + +# --------------------------------------------------------------------------- +# Exceptions +# --------------------------------------------------------------------------- + + +class RiskGuardError(Exception): + """Raised when a risk guard blocks a proposed trade or portfolio action.""" + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def kelly_fraction(win_rate: float, payoff_ratio: float) -> float: + """Compute the full Kelly fraction. + + Formula: f* = W − (1−W) / R + + Args: + win_rate: historical win rate, strictly in (0, 1). + payoff_ratio: average-win / average-loss ratio, > 0. + + Returns: + Raw Kelly fraction; may be ≤ 0 (no positive edge). + + Raises: + ValueError: if arguments are out of their valid ranges. + """ + if not (0.0 < win_rate < 1.0): + raise ValueError(f"win_rate must be in (0, 1); got {win_rate!r}") + if payoff_ratio <= 0.0: + raise ValueError(f"payoff_ratio must be > 0; got {payoff_ratio!r}") + return win_rate - (1.0 - win_rate) / payoff_ratio + + +def fractional_kelly( + win_rate: float, + payoff_ratio: float, + fraction: float = KELLY_DEFAULT_FRACTION, +) -> float: + """Return the fractional-Kelly equity fraction to risk. + + Clamps the multiplier to ≤ ``KELLY_MAX_FRACTION`` (half-Kelly) and + returns 0.0 when there is no positive edge. + + Args: + win_rate: historical win rate. + payoff_ratio: average-win / average-loss. + fraction: Kelly multiplier (default 0.25; capped at 0.5). + + Returns: + Fraction of equity to risk; always in [0.0, ``KELLY_MAX_FRACTION``]. + """ + fraction = min(fraction, KELLY_MAX_FRACTION) + f_star = kelly_fraction(win_rate, payoff_ratio) + if f_star <= 0.0: + return 0.0 + return min(fraction * f_star, KELLY_MAX_FRACTION) + + +def compute_position_size( + *, + portfolio: Portfolio, + instrument: str, + entry_price: float, + stop_price: float, + risk_pct: float = RISK_PCT_DEFAULT, + win_rate: float | None = None, + payoff_ratio: float | None = None, + kelly_fraction_override: float = KELLY_DEFAULT_FRACTION, +) -> SizeResult: + """Compute the allowed position size subject to all active risk guards. + + Guard hierarchy (applied in order): + 1. Circuit-breaker — block trade (HALTED) or halve risk budget (HALVED). + 2. Per-trade risk cap — equity × effective_risk_pct. + 3. Fractional-Kelly cap — if win_rate and payoff_ratio are supplied. + 4. Take the *smaller* of risk-cap and Kelly-derived size. + 5. Portfolio-heat guard — trim to remaining heat budget. + 6. Single-name concentration cap — trim to remaining single-name budget. + 7. Correlation cap — hard block if already at the maximum. + + Args: + portfolio: current portfolio state. + instrument: ticker / instrument identifier. + entry_price: intended entry price. + stop_price: mechanical stop price. Required — no stop, no trade. + risk_pct: per-trade risk fraction (default 0.01; clamped to + [RISK_PCT_MIN, RISK_PCT_MAX]). + win_rate: historical win rate for Kelly sizing; None skips Kelly. + payoff_ratio: average-win / average-loss; required when win_rate given. + kelly_fraction_override: Kelly multiplier (default 0.25; hard cap 0.5). + + Returns: + :class:`SizeResult` with final share count and decision metadata. + + Raises: + RiskGuardError: when a hard guard blocks the trade entirely. + ValueError: for invalid argument values. + """ + if entry_price <= 0: + raise ValueError(f"entry_price must be > 0; got {entry_price!r}") + risk_per_share = abs(entry_price - stop_price) + if risk_per_share == 0.0: + raise RiskGuardError("stop_price equals entry_price — no stop defined; trade rejected") + + # 1. Circuit-breaker ------------------------------------------------ + cb_state = portfolio.circuit_breaker_state + if cb_state == CircuitBreakerState.HALTED: + raise RiskGuardError( + f"Trading halted: drawdown {portfolio.drawdown:.1%} reached the " + f"{DD_HALT_THRESHOLD:.0%} threshold. Flatten all positions and stop." + ) + + # 2. Clamp risk_pct, apply halve-modifier + risk_pct = max(RISK_PCT_MIN, min(risk_pct, RISK_PCT_MAX)) + effective_risk_pct = risk_pct * (0.5 if cb_state == CircuitBreakerState.HALVED else 1.0) + equity = portfolio.equity + + risk_dollars_cap = equity * effective_risk_pct + shares_from_cap = risk_dollars_cap / risk_per_share + + # 3. Fractional-Kelly sizing ---------------------------------------- + kelly_frac_used = 0.0 + shares_from_kelly: float | None = None + if win_rate is not None: + if payoff_ratio is None: + raise ValueError("payoff_ratio is required when win_rate is supplied") + kelly_frac_used = fractional_kelly(win_rate, payoff_ratio, kelly_fraction_override) + risk_dollars_kelly = equity * kelly_frac_used + shares_from_kelly = risk_dollars_kelly / risk_per_share + + # 4. Take the smaller ------------------------------------------------- + if shares_from_kelly is not None and shares_from_kelly < shares_from_cap: + shares = shares_from_kelly + risk_dollars = equity * kelly_frac_used + binding_rule = "kelly" + else: + shares = shares_from_cap + risk_dollars = risk_dollars_cap + binding_rule = "risk_cap" + + if shares <= 0.0: + raise RiskGuardError( + "Computed position size is zero — no positive edge or stop too tight" + ) + + # 5. Portfolio heat guard ------------------------------------------- + existing_heat = portfolio.total_open_risk + new_heat_pct = (existing_heat + risk_dollars) / equity + if new_heat_pct > PORTFOLIO_HEAT_LIMIT: + available_dollars = max(0.0, PORTFOLIO_HEAT_LIMIT * equity - existing_heat) + if available_dollars <= 0.0: + raise RiskGuardError( + f"Portfolio heat limit ({PORTFOLIO_HEAT_LIMIT:.0%}) already reached; " + "no new risk capacity available." + ) + shares = available_dollars / risk_per_share + risk_dollars = available_dollars + binding_rule = "heat_limit" + + # 6. Single-name concentration cap ---------------------------------- + existing_exposure = portfolio.exposure_for(instrument) + new_exposure = existing_exposure + entry_price * shares + if new_exposure / equity > CONCENTRATION_CAP: + allowed_exposure = CONCENTRATION_CAP * equity - existing_exposure + if allowed_exposure <= 0.0: + raise RiskGuardError( + f"Concentration cap ({CONCENTRATION_CAP:.0%}) already reached " + f"for {instrument!r}." + ) + shares = allowed_exposure / entry_price + risk_dollars = shares * risk_per_share + binding_rule = "concentration" + + # 7. Correlation cap ------------------------------------------------ + corr_count = portfolio.correlated_position_count(instrument) + if corr_count >= CORRELATION_MAX_POSITIONS: + raise RiskGuardError( + f"Correlation cap: {corr_count} existing position(s) with ρ > " + f"{CORRELATION_THRESHOLD} already open relative to {instrument!r}; " + f"max is {CORRELATION_MAX_POSITIONS}." + ) + + return SizeResult( + shares=shares, + risk_dollars=risk_dollars, + kelly_fraction_used=kelly_frac_used, + binding_rule=binding_rule, + cb_state=cb_state, + ) + + +def check_portfolio_heat(portfolio: Portfolio) -> None: + """Assert current portfolio heat is within policy limits. + + Useful as a standalone pre-trade check independent of position sizing. + + Raises: + RiskGuardError: if total open risk already exceeds the heat limit. + ValueError: if equity is not positive. + """ + if portfolio.equity <= 0: + raise ValueError("equity must be positive") + heat_pct = portfolio.total_open_risk_pct + if heat_pct > PORTFOLIO_HEAT_LIMIT: + raise RiskGuardError( + f"Portfolio heat {heat_pct:.2%} exceeds the " + f"{PORTFOLIO_HEAT_LIMIT:.0%} limit." + ) diff --git a/backend/tests/test_risk_guard.py b/backend/tests/test_risk_guard.py new file mode 100644 index 0000000..18d5121 --- /dev/null +++ b/backend/tests/test_risk_guard.py @@ -0,0 +1,499 @@ +"""Unit tests for app.services.risk_guard. + +Coverage targets: + - kelly_fraction formula correctness + - fractional_kelly: default quarter-Kelly, half-Kelly cap, no negative output + - compute_position_size: risk-cap sizing, Kelly sizing, binding-rule selection + - Drawdown circuit-breaker: NORMAL / HALVED / HALTED states + - Portfolio heat guard: trim and hard block + - Single-name concentration cap: trim and hard block + - Correlation cap: block at limit + - Edge cases: zero risk_per_share, equity = 0 in heat check +""" + +from __future__ import annotations + +import pytest + +from app.services.risk_guard import ( + CORRELATION_MAX_POSITIONS, + CORRELATION_THRESHOLD, + DD_HALT_THRESHOLD, + DD_HALVE_THRESHOLD, + KELLY_DEFAULT_FRACTION, + KELLY_MAX_FRACTION, + PORTFOLIO_HEAT_LIMIT, + RISK_PCT_DEFAULT, + CircuitBreakerState, + Portfolio, + Position, + RiskGuardError, + SizeResult, + check_portfolio_heat, + compute_position_size, + fractional_kelly, + kelly_fraction, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _clean_portfolio(equity: float = 100_000.0) -> Portfolio: + """Return an empty, healthy portfolio with equity == peak_equity.""" + return Portfolio(equity=equity, peak_equity=equity) + + +def _sized_position( + instrument: str = "AAPL", + entry: float = 100.0, + stop: float = 95.0, + shares: float = 10.0, + equity: float = 100_000.0, +) -> Position: + return Position( + instrument=instrument, + entry_price=entry, + stop_price=stop, + shares=shares, + equity_at_entry=equity, + ) + + +# --------------------------------------------------------------------------- +# kelly_fraction +# --------------------------------------------------------------------------- + + +def test_kelly_fraction_breakeven_edge() -> None: + # W=0.5, R=1.0 → f* = 0.5 - 0.5/1.0 = 0.0 + assert kelly_fraction(0.5, 1.0) == pytest.approx(0.0) + + +def test_kelly_fraction_positive_edge() -> None: + # W=0.6, R=2.0 → f* = 0.6 - 0.4/2.0 = 0.6 - 0.2 = 0.4 + assert kelly_fraction(0.6, 2.0) == pytest.approx(0.4) + + +def test_kelly_fraction_negative_edge() -> None: + # W=0.3, R=1.0 → f* = 0.3 - 0.7 = -0.4 (no edge) + assert kelly_fraction(0.3, 1.0) == pytest.approx(-0.4) + + +def test_kelly_fraction_bad_win_rate() -> None: + with pytest.raises(ValueError, match="win_rate"): + kelly_fraction(0.0, 2.0) + with pytest.raises(ValueError, match="win_rate"): + kelly_fraction(1.0, 2.0) + + +def test_kelly_fraction_bad_payoff() -> None: + with pytest.raises(ValueError, match="payoff_ratio"): + kelly_fraction(0.5, 0.0) + + +# --------------------------------------------------------------------------- +# fractional_kelly +# --------------------------------------------------------------------------- + + +def test_fractional_kelly_default_quarter() -> None: + # f* = 0.4; quarter-Kelly → 0.1 + result = fractional_kelly(0.6, 2.0) + assert result == pytest.approx(0.4 * KELLY_DEFAULT_FRACTION) + + +def test_fractional_kelly_no_edge_returns_zero() -> None: + result = fractional_kelly(0.3, 1.0) # f* < 0 + assert result == 0.0 + + +def test_fractional_kelly_capped_at_half_kelly() -> None: + # Passing fraction=1.0 (full Kelly) must be clamped down to KELLY_MAX_FRACTION=0.5. + # Since f* < 1.0 always, the final result is KELLY_MAX_FRACTION × f*, not + # KELLY_MAX_FRACTION itself — but it must equal what you'd get with fraction=0.5. + result_full = fractional_kelly(0.6, 2.0, fraction=1.0) + result_half = fractional_kelly(0.6, 2.0, fraction=KELLY_MAX_FRACTION) + assert result_full == pytest.approx(result_half) + # And the result must never exceed KELLY_MAX_FRACTION + assert result_full <= KELLY_MAX_FRACTION + + +def test_fractional_kelly_respects_max_fraction() -> None: + # fraction arg already above cap — must be clamped + result = fractional_kelly(0.9, 5.0, fraction=0.9) + assert result <= KELLY_MAX_FRACTION + + +# --------------------------------------------------------------------------- +# compute_position_size — basic risk-cap sizing +# --------------------------------------------------------------------------- + + +def test_risk_cap_binding() -> None: + # equity=100k, risk_pct=1%, entry=100, stop=95 → risk$=1000, shares=200 + portfolio = _clean_portfolio() + result = compute_position_size( + portfolio=portfolio, + instrument="AAPL", + entry_price=100.0, + stop_price=95.0, + ) + assert result.binding_rule == "risk_cap" + assert result.shares == pytest.approx(200.0) + assert result.risk_dollars == pytest.approx(1_000.0) + assert result.cb_state == CircuitBreakerState.NORMAL + + +def test_kelly_beats_risk_cap() -> None: + # f*=0.4, quarter→0.1 → risk$=10k, shares=2000 > 200 from risk cap + # risk_cap (1%) = 1000 → 200 shares; Kelly (10%) = 10000 → 2000 shares + # risk cap is smaller → binding_rule should be risk_cap + portfolio = _clean_portfolio() + result = compute_position_size( + portfolio=portfolio, + instrument="AAPL", + entry_price=100.0, + stop_price=95.0, + win_rate=0.6, + payoff_ratio=2.0, + ) + # Kelly would give 2000 shares; risk cap gives 200 — risk cap wins + assert result.binding_rule == "risk_cap" + assert result.shares == pytest.approx(200.0) + + +def test_kelly_smaller_than_risk_cap() -> None: + # Force Kelly to be binding: high risk_pct + small Kelly fraction + # risk_pct=3%, equity=100k → cap=3000/share; Kelly: f*=0.4 at 0.05 frac → 0.02 → 2000 + portfolio = _clean_portfolio(equity=100_000.0) + result = compute_position_size( + portfolio=portfolio, + instrument="SPY", + entry_price=100.0, + stop_price=90.0, # risk_per_share=10 + risk_pct=0.03, # 3k risk → 300 shares from cap + win_rate=0.6, + payoff_ratio=2.0, + kelly_fraction_override=0.05, # very small Kelly fraction → 0.02 of equity → 2k → 200 shares + ) + # Kelly: 0.4 * 0.05 = 0.02 of equity → 2000$ → 200 shares + # Cap: 3% of 100k = 3000$ → 300 shares + # Kelly is smaller → kelly binding + assert result.binding_rule == "kelly" + assert result.shares == pytest.approx(200.0) + + +def test_no_stop_raises() -> None: + portfolio = _clean_portfolio() + with pytest.raises(RiskGuardError, match="no stop"): + compute_position_size( + portfolio=portfolio, + instrument="AAPL", + entry_price=100.0, + stop_price=100.0, + ) + + +def test_invalid_entry_price() -> None: + portfolio = _clean_portfolio() + with pytest.raises(ValueError, match="entry_price"): + compute_position_size( + portfolio=portfolio, + instrument="AAPL", + entry_price=0.0, + stop_price=95.0, + ) + + +def test_win_rate_without_payoff_raises() -> None: + portfolio = _clean_portfolio() + with pytest.raises(ValueError, match="payoff_ratio"): + compute_position_size( + portfolio=portfolio, + instrument="AAPL", + entry_price=100.0, + stop_price=95.0, + win_rate=0.6, + # payoff_ratio omitted + ) + + +# --------------------------------------------------------------------------- +# Circuit-breaker states +# --------------------------------------------------------------------------- + + +def test_circuit_breaker_normal() -> None: + # 10% drawdown: NORMAL + p = Portfolio(equity=90_000.0, peak_equity=100_000.0) + assert p.circuit_breaker_state == CircuitBreakerState.NORMAL + assert p.drawdown == pytest.approx(-0.10) + + +def test_circuit_breaker_halved() -> None: + # −20 % drawdown: HALVED + p = Portfolio(equity=80_000.0, peak_equity=100_000.0) + assert p.circuit_breaker_state == CircuitBreakerState.HALVED + + +def test_circuit_breaker_halted() -> None: + # −30 % drawdown: HALTED + p = Portfolio(equity=70_000.0, peak_equity=100_000.0) + assert p.circuit_breaker_state == CircuitBreakerState.HALTED + + +def test_halted_circuit_breaker_blocks_trade() -> None: + p = Portfolio(equity=70_000.0, peak_equity=100_000.0) + with pytest.raises(RiskGuardError, match="halted|halve|flatten"): + compute_position_size( + portfolio=p, + instrument="AAPL", + entry_price=100.0, + stop_price=95.0, + ) + + +def test_halved_circuit_breaker_cuts_size_in_half() -> None: + # −20 % drawdown: effective risk_pct should be 0.5 % + p = Portfolio(equity=80_000.0, peak_equity=100_000.0) + result = compute_position_size( + portfolio=p, + instrument="AAPL", + entry_price=100.0, + stop_price=95.0, # risk_per_share=5 + ) + # effective_risk_pct = 0.01 * 0.5 = 0.005; risk$ = 80k * 0.005 = 400 + # shares = 400 / 5 = 80 + assert result.cb_state == CircuitBreakerState.HALVED + assert result.shares == pytest.approx(80.0) + assert result.risk_dollars == pytest.approx(400.0) + + +# --------------------------------------------------------------------------- +# Portfolio heat guard +# --------------------------------------------------------------------------- + + +def test_portfolio_heat_trims_size() -> None: + # Fill 5% heat, try to add 2% → trimmed to 1% + equity = 100_000.0 + existing_risk = equity * 0.05 # 5 000$ + # One position: entry=100, stop=90, risk_per_share=10 + shares_existing = existing_risk / 10.0 + p = Portfolio( + equity=equity, + peak_equity=equity, + open_positions=[ + Position("X", 100.0, 90.0, shares_existing, equity) + ], + ) + result = compute_position_size( + portfolio=p, + instrument="Y", + entry_price=50.0, + stop_price=40.0, # risk_per_share=10 + risk_pct=0.02, + ) + assert result.binding_rule == "heat_limit" + # Available heat = (6% - 5%) × 100k = 1000$; shares = 1000/10 = 100 + assert result.risk_dollars == pytest.approx(1_000.0) + assert result.shares == pytest.approx(100.0) + + +def test_portfolio_heat_full_blocks_trade() -> None: + equity = 100_000.0 + risk_per_share = 5.0 + # 6% of equity as existing risk + shares_existing = (equity * PORTFOLIO_HEAT_LIMIT) / risk_per_share + p = Portfolio( + equity=equity, + peak_equity=equity, + open_positions=[Position("Z", 100.0, 95.0, shares_existing, equity)], + ) + with pytest.raises(RiskGuardError, match="heat limit"): + compute_position_size( + portfolio=p, + instrument="NEW", + entry_price=100.0, + stop_price=95.0, + ) + + +def test_check_portfolio_heat_ok() -> None: + p = _clean_portfolio() + check_portfolio_heat(p) # no exception + + +def test_check_portfolio_heat_exceeded() -> None: + equity = 100_000.0 + p = Portfolio( + equity=equity, + peak_equity=equity, + open_positions=[Position("A", 100.0, 90.0, 700.0, equity)], + ) + # 700 * 10 = 7000 = 7% > 6% + with pytest.raises(RiskGuardError, match="heat"): + check_portfolio_heat(p) + + +def test_check_portfolio_heat_bad_equity() -> None: + p = Portfolio(equity=0.0, peak_equity=0.0) + with pytest.raises(ValueError, match="equity"): + check_portfolio_heat(p) + + +# --------------------------------------------------------------------------- +# Single-name concentration cap +# --------------------------------------------------------------------------- + + +def test_concentration_cap_trims_size() -> None: + equity = 100_000.0 + # Existing 15% exposure to AAPL: 15k at entry=100 → 150 shares + existing_shares = (equity * 0.15) / 100.0 + p = Portfolio( + equity=equity, + peak_equity=equity, + open_positions=[Position("AAPL", 100.0, 90.0, existing_shares, equity)], + ) + # Requesting another batch of AAPL; cap is 20% → only 5k more allowed + result = compute_position_size( + portfolio=p, + instrument="AAPL", + entry_price=100.0, + stop_price=90.0, + risk_pct=0.03, + ) + # Max additional exposure = 5k → 50 shares + assert result.binding_rule == "concentration" + assert result.shares == pytest.approx(50.0, rel=1e-4) + + +def test_concentration_cap_hard_block() -> None: + equity = 100_000.0 + # Already at 20% exposure + existing_shares = (equity * 0.20) / 100.0 + p = Portfolio( + equity=equity, + peak_equity=equity, + open_positions=[Position("TSLA", 100.0, 90.0, existing_shares, equity)], + ) + with pytest.raises(RiskGuardError, match="[Cc]oncentration"): + compute_position_size( + portfolio=p, + instrument="TSLA", + entry_price=100.0, + stop_price=90.0, + ) + + +# --------------------------------------------------------------------------- +# Correlation cap +# --------------------------------------------------------------------------- + + +def test_correlation_cap_blocks_third_correlated_position() -> None: + equity = 100_000.0 + # Two existing positions both highly correlated with the new instrument + p = Portfolio( + equity=equity, + peak_equity=equity, + open_positions=[ + Position("SPY", 400.0, 390.0, 5.0, equity), + Position("IVV", 401.0, 391.0, 5.0, equity), + ], + correlations={ + frozenset({"QQQ", "SPY"}): 0.92, + frozenset({"QQQ", "IVV"}): 0.88, + }, + ) + with pytest.raises(RiskGuardError, match="[Cc]orrelation"): + compute_position_size( + portfolio=p, + instrument="QQQ", + entry_price=350.0, + stop_price=340.0, + ) + + +def test_correlation_cap_allows_second_correlated_position() -> None: + equity = 100_000.0 + # Only one existing correlated position → still allowed + p = Portfolio( + equity=equity, + peak_equity=equity, + open_positions=[ + Position("SPY", 400.0, 390.0, 5.0, equity), + ], + correlations={ + frozenset({"QQQ", "SPY"}): 0.92, + }, + ) + result = compute_position_size( + portfolio=p, + instrument="QQQ", + entry_price=350.0, + stop_price=340.0, + ) + assert isinstance(result, SizeResult) + + +def test_low_correlation_not_counted() -> None: + equity = 100_000.0 + p = Portfolio( + equity=equity, + peak_equity=equity, + open_positions=[ + Position("GLD", 200.0, 190.0, 5.0, equity), + Position("BTC", 60_000.0, 58_000.0, 0.01, equity), + ], + correlations={ + frozenset({"AAPL", "GLD"}): 0.10, + frozenset({"AAPL", "BTC"}): 0.35, + }, + ) + # Neither is above CORRELATION_THRESHOLD → count = 0 → allowed + result = compute_position_size( + portfolio=p, + instrument="AAPL", + entry_price=150.0, + stop_price=145.0, + ) + assert isinstance(result, SizeResult) + + +# --------------------------------------------------------------------------- +# risk_pct clamping +# --------------------------------------------------------------------------- + + +def test_risk_pct_clamped_to_max() -> None: + # risk_pct=10% should be silently clamped to 3%. + # Use a low entry price so the concentration cap (20% of equity) doesn't bind: + # 600 shares × $10 = $6 000 = 6 % of $100 k — well under the 20 % cap. + portfolio = _clean_portfolio() + result = compute_position_size( + portfolio=portfolio, + instrument="AAPL", + entry_price=10.0, + stop_price=5.0, # risk_per_share = 5 + risk_pct=0.10, + ) + # Clamped to 3 %: 3 000 / 5 = 600 shares + assert result.binding_rule == "risk_cap" + assert result.shares == pytest.approx(600.0) + + +def test_risk_pct_clamped_to_min() -> None: + # risk_pct=0.001% should be clamped up to 1% + portfolio = _clean_portfolio() + result = compute_position_size( + portfolio=portfolio, + instrument="AAPL", + entry_price=100.0, + stop_price=95.0, + risk_pct=0.0001, + ) + assert result.shares == pytest.approx(200.0) # 1% of 100k / 5