"""Risk guardrails (requirements §8).
Checked here and never inside a strategy or evaluator. The app enforces limits
on the thing being limited, rather than asking it to limit itself — the same
principle as the Slice 1 safety gate.
"""
import logging
from dataclasses import dataclass
from decimal import Decimal
from trader.config.schema import PipelineConfig
from trader.domain import Account
__all__ = [
"REASON_DAILY_LOSS_HALT",
"REASON_DISCOVERY_CAP",
"REASON_INVALID_PRICE",
"REASON_POSITION_CAP",
"REASON_TOTAL_EXPOSURE_CAP",
"ExposureCounter",
"Guardrails",
"RiskDecision",
]
_HUNDRED = Decimal(100)
_logger = logging.getLogger("trader.risk")
# One structured code per refusal `check_entry` (or `daily_loss_breached`, via
# `check_entry`) can produce. `decisions.rejection_reason` (issue #21) stores
# exactly one of these — never `RiskDecision.reason`'s free text, which is for
# a human. `run_once.py` imports `REASON_DAILY_LOSS_HALT` directly, because its
# own halted check (`_process_symbol`) short-circuits before `check_entry` is
# ever reached and must name the same gate rather than inventing a second
# string for the same fact.
REASON_DAILY_LOSS_HALT = "daily_loss_halt"
REASON_DISCOVERY_CAP = "discovery_cap"
REASON_INVALID_PRICE = "invalid_price"
REASON_TOTAL_EXPOSURE_CAP = "total_exposure_cap"
REASON_POSITION_CAP = "position_cap"
[docs]
@dataclass(slots=True)
class ExposureCounter:
"""A mutable running total of capital committed this cycle.
Seeded before the symbol loop from open positions plus already-pending
entry orders, then advanced by `commit()` every time an entry is actually
submitted — so the next symbol in the same cycle sees the reduced
headroom.
Mutable on purpose. A total computed once and never updated is blind to
what the cycle itself is spending, which is precisely how twelve limit
buys left one cycle on 2026-08-04, and the same mistake the
discovered-position cap made before it.
"""
committed: Decimal
[docs]
def commit(self, amount: Decimal) -> None:
"""Record capital just committed by a submitted entry."""
self.committed += amount
[docs]
@dataclass(frozen=True, slots=True)
class RiskDecision:
"""The outcome of a guardrail check, and why.
A rejection is not an error — it is a recorded Decision explaining why the
trade did not happen. `reason` is therefore populated on every path,
including approval.
`code` is one of the module's `REASON_*` constants on a refusal
(`allowed=False`), and `None` on every approval path — including a shrink,
which is not a refusal. `decisions.rejection_reason` (issue #21) is
written from this, never from `reason`'s free text.
"""
allowed: bool
reason: str
approved_quantity: int
code: str | None = None
[docs]
class Guardrails:
"""Position-size and daily-loss limits."""
def __init__(self, config: PipelineConfig) -> None:
self._config = config
[docs]
def daily_loss_breached(self, account: Account) -> bool:
"""Whether today's drawdown has reached the configured limit.
Uses the broker's own `equity` and `last_equity` rather than
reconstructing realized P/L from trade history, which would have to
model partial fills and corporate actions to arrive at the same number.
"""
if account.last_equity <= 0:
# A new or unfunded account. No baseline means no meaningful
# percentage, and halting on that basis would be arbitrary.
return False
change_pct = (
(account.equity - account.last_equity) / account.last_equity
) * _HUNDRED
# `<=`, not `<`: a limit that is reached is a limit that is hit.
return change_pct <= -self._config.daily_loss_limit_pct
[docs]
def check_entry(
self,
account: Account,
symbol: str,
proposed_quantity: int,
price: Decimal,
*,
is_discovered: bool = False,
discovered_positions_held: int = 0,
max_discovered_positions: int = 0,
exposure: "ExposureCounter | None" = None,
) -> RiskDecision:
"""Approve, shrink, or reject a proposed entry.
The discovery arguments are keyword-only with defaults so that every
pre-Slice-3d caller behaves exactly as before: a fixed ticker is never
subject to the discovered cap.
"""
if self.daily_loss_breached(account):
reason = (
f"Daily loss limit of {self._config.daily_loss_limit_pct}% reached "
f"(equity {account.equity} against {account.last_equity} at "
"yesterday's close). No new entries for the rest of the day; "
"existing protective stops are untouched."
)
_logger.warning("guardrail halt %s: %s", symbol, reason)
return RiskDecision(
allowed=False,
reason=reason,
approved_quantity=0,
code=REASON_DAILY_LOSS_HALT,
)
if is_discovered and discovered_positions_held >= max_discovered_positions:
reason = (
f"Already holding {discovered_positions_held} discovered "
f"position(s), at the limit of {max_discovered_positions}. "
f"{symbol} was found by discovery, not chosen in "
"fixed_tickers, so it waits."
)
_logger.info("guardrail discovered-cap %s: %s", symbol, reason)
return RiskDecision(
allowed=False,
reason=reason,
approved_quantity=0,
code=REASON_DISCOVERY_CAP,
)
if price <= 0:
return RiskDecision(
allowed=False,
reason=(f"Refusing to size an order for {symbol} at a price of {price}."),
approved_quantity=0,
code=REASON_INVALID_PRICE,
)
# Measured against portfolio value, not cash: the cap is about
# concentration in one name, and cash fluctuates with open positions.
cap = account.portfolio_value * self._config.max_position_pct / _HUNDRED
affordable = int(cap // price)
# The aggregate ceiling, applied alongside the per-name one so the
# tighter of the two wins. `max_position_pct` never bounded the SUM:
# eight fixed tickers plus three discovered slots at 10% each is 110%
# of equity, and Alpaca's paper margin would happily fund it.
if exposure is not None:
total_cap = (
account.portfolio_value * self._config.max_total_exposure_pct / _HUNDRED
)
headroom = total_cap - exposure.committed
by_exposure = int(headroom // price) if headroom > 0 else 0
if by_exposure < 1:
reason = (
f"Total exposure limit of "
f"{self._config.max_total_exposure_pct}% "
f"({total_cap} of {account.portfolio_value}) is already "
f"committed ({exposure.committed}); no room for {symbol} "
f"at {price}."
)
_logger.info("guardrail exposure %s: %s", symbol, reason)
return RiskDecision(
allowed=False,
reason=reason,
approved_quantity=0,
code=REASON_TOTAL_EXPOSURE_CAP,
)
affordable = min(affordable, by_exposure)
if affordable < 1:
reason = (
f"Max position size of {self._config.max_position_pct}% of "
f"{account.portfolio_value} is {cap}, which does not cover one "
f"share of {symbol} at {price}."
)
_logger.info("guardrail reject %s: %s", symbol, reason)
return RiskDecision(
allowed=False,
reason=reason,
approved_quantity=0,
code=REASON_POSITION_CAP,
)
if affordable < proposed_quantity:
reason = (
f"Shrunk from {proposed_quantity} to {affordable} shares by the "
f"max position size of {self._config.max_position_pct}% "
f"({cap} of {account.portfolio_value})."
)
_logger.info("guardrail shrink %s: %s", symbol, reason)
return RiskDecision(allowed=True, reason=reason, approved_quantity=affordable)
return RiskDecision(
allowed=True,
reason=(
f"{proposed_quantity} shares at {price} is within the "
f"{self._config.max_position_pct}% position cap of {cap}."
),
approved_quantity=proposed_quantity,
)