diff --git a/services/trade_executor/config.py b/services/trade_executor/config.py
index 49df392..1b7b1a5 100644
--- a/services/trade_executor/config.py
+++ b/services/trade_executor/config.py
@@ -15,4 +15,11 @@ class TradeExecutorConfig(BaseConfig):
     alpaca_secret_key: str = ""
     paper_trading: bool = True
 
+    # Kevin v2 risk caps — only applied when TradeSignal.strategy_id ==
+    # KEVIN_STRATEGY_UUID.
+    kevin_daily_trade_cap: int = 10
+    kevin_daily_alloc_cap_usd: float = 20_000.0
+    kevin_equity_drawdown_halt_pct: float = 0.20  # 20% drawdown → permanent pause
+    kevin_daily_loss_circuit_pct: float = 0.05  # 5% daily loss → 24h pause
+
     model_config = {"env_prefix": "TRADING_"}
diff --git a/services/trade_executor/risk_manager.py b/services/trade_executor/risk_manager.py
index dcfeffa..238dfbe 100644
--- a/services/trade_executor/risk_manager.py
+++ b/services/trade_executor/risk_manager.py
@@ -8,12 +8,14 @@ from __future__ import annotations
 
 import logging
 from datetime import datetime, timedelta
+from decimal import Decimal
 from zoneinfo import ZoneInfo
 
 from redis.asyncio import Redis
 
 from services.trade_executor.config import TradeExecutorConfig
 from shared.broker.base import BaseBroker
+from shared.constants.kevin import KEVIN_STRATEGY_UUID
 from shared.schemas.trading import AccountInfo, PositionInfo, SignalDirection, TradeSignal
 
 logger = logging.getLogger(__name__)
@@ -28,6 +30,13 @@ _MARKET_CLOSE_MINUTE = 0
 
 TRADING_PAUSED_KEY = "trading:paused"
 
+# Kevin counter keys — _today_key() suffixes with YYYY-MM-DD in ET.
+_KEVIN_DAILY_TRADES_PREFIX = "kevin:daily_trades:"
+_KEVIN_DAILY_ALLOC_PREFIX = "kevin:daily_alloc_usd:"
+_KEVIN_DAILY_PNL_PREFIX = "kevin:daily_pnl_usd:"
+_KEVIN_STARTING_EQUITY_KEY = "kevin:starting_equity_usd"
+_PAUSE_24H_SECONDS = 86400
+
 
 class RiskManager:
     """Performs pre-trade risk checks and calculates position sizes.
@@ -98,6 +107,75 @@ class RiskManager:
         if total_exposure >= max_exposure:
             return False, "max_exposure_exceeded"
 
+        # 5. Kevin-specific caps (only when the signal carries the Kevin
+        #    strategy ID; no-op for other strategies).
+        if signal.strategy_id == KEVIN_STRATEGY_UUID and self.redis is not None:
+            kevin_ok, kevin_reason = await self._check_kevin_caps(
+                signal, account, now_et
+            )
+            if not kevin_ok:
+                return False, kevin_reason
+
+        return True, "approved"
+
+    async def _check_kevin_caps(
+        self,
+        signal: TradeSignal,
+        account: AccountInfo,
+        now_et: datetime,
+    ) -> tuple[bool, str]:
+        """Apply the 4 Kevin v2 risk caps. Caller already verified
+        ``signal.strategy_id == KEVIN_STRATEGY_UUID`` and ``self.redis``
+        is not None.
+        """
+        today = now_et.strftime("%Y-%m-%d")
+
+        # 5a. Drawdown halt (permanent) — checks BEFORE the daily caps so a
+        #     blown-up account stops trading immediately.
+        starting_equity_raw = await self.redis.get(_KEVIN_STARTING_EQUITY_KEY)
+        if starting_equity_raw is not None:
+            starting_equity = float(starting_equity_raw)
+            if starting_equity > 0:
+                drawdown = (starting_equity - account.equity) / starting_equity
+                if drawdown >= self.config.kevin_equity_drawdown_halt_pct:
+                    logger.warning(
+                        "Kevin drawdown halt: equity %.2f vs starting %.2f (%.1f%%)",
+                        account.equity,
+                        starting_equity,
+                        drawdown * 100,
+                    )
+                    await self.redis.set(TRADING_PAUSED_KEY, "1")
+                    return False, "kevin_drawdown_halt"
+
+        # 5b. Daily-loss circuit breaker (24h pause).
+        daily_pnl_raw = await self.redis.get(_KEVIN_DAILY_PNL_PREFIX + today)
+        if daily_pnl_raw is not None:
+            daily_pnl_usd = float(daily_pnl_raw)
+            loss_threshold_usd = -account.equity * self.config.kevin_daily_loss_circuit_pct
+            if daily_pnl_usd <= loss_threshold_usd:
+                logger.warning(
+                    "Kevin daily-loss circuit: today P&L %.2f <= threshold %.2f",
+                    daily_pnl_usd,
+                    loss_threshold_usd,
+                )
+                await self.redis.setex(TRADING_PAUSED_KEY, _PAUSE_24H_SECONDS, "1")
+                return False, "kevin_daily_loss_circuit"
+
+        # 5c. Daily trade count cap.
+        daily_trades_raw = await self.redis.get(_KEVIN_DAILY_TRADES_PREFIX + today)
+        daily_trades = int(daily_trades_raw) if daily_trades_raw is not None else 0
+        if daily_trades >= self.config.kevin_daily_trade_cap:
+            return False, "kevin_daily_trade_cap"
+
+        # 5d. Daily allocation cap (allocated $ today + this trade's notional).
+        daily_alloc_raw = await self.redis.get(_KEVIN_DAILY_ALLOC_PREFIX + today)
+        daily_alloc = float(daily_alloc_raw) if daily_alloc_raw is not None else 0.0
+        this_trade_usd = (
+            float(signal.target_dollars) if signal.target_dollars is not None else 0.0
+        )
+        if daily_alloc + this_trade_usd > self.config.kevin_daily_alloc_cap_usd:
+            return False, "kevin_daily_alloc_cap"
+
         return True, "approved"
 
     def calculate_position_size(
diff --git a/shared/broker/alpaca_broker.py b/shared/broker/alpaca_broker.py
index 3e2fd42..fbaaad6 100644
--- a/shared/broker/alpaca_broker.py
+++ b/shared/broker/alpaca_broker.py
@@ -14,6 +14,7 @@ from decimal import Decimal
 
 from alpaca.common.exceptions import APIError
 from alpaca.trading.client import TradingClient
+from alpaca.trading.enums import OrderClass as AlpacaOrderClass
 from alpaca.trading.enums import OrderSide as AlpacaOrderSide
 from alpaca.trading.enums import OrderStatus as AlpacaOrderStatus
 from alpaca.trading.enums import TimeInForce
@@ -23,7 +24,9 @@ from alpaca.trading.models import TradeAccount
 from alpaca.trading.requests import (
     LimitOrderRequest,
     MarketOrderRequest,
+    StopLossRequest,
     StopOrderRequest,
+    TakeProfitRequest,
 )
 
 from shared.broker.base import BaseBroker
@@ -110,6 +113,23 @@ class AlpacaBroker(BaseBroker):
         """Convert our ``OrderRequest`` into the appropriate Alpaca request."""
         side = AlpacaOrderSide.BUY if order.side == OrderSide.BUY else AlpacaOrderSide.SELL
 
+        if order.order_class == "bracket":
+            # Bracket only attaches to MARKET parent legs in the Kevin path
+            # (entry on signal, stop + take-profit live on the broker).
+            return MarketOrderRequest(
+                symbol=order.ticker,
+                qty=order.qty,
+                side=side,
+                time_in_force=TimeInForce.GTC,
+                order_class=AlpacaOrderClass.BRACKET,
+                take_profit=TakeProfitRequest(
+                    limit_price=order.take_profit_price,
+                ),
+                stop_loss=StopLossRequest(
+                    stop_price=order.stop_loss_price,
+                ),
+            )
+
         if order.order_type == OrderType.LIMIT:
             if order.limit_price is None:
                 raise ValueError("limit_price is required for limit orders")
diff --git a/shared/schemas/trading.py b/shared/schemas/trading.py
index 56cc751..5c8c7d0 100644
--- a/shared/schemas/trading.py
+++ b/shared/schemas/trading.py
@@ -3,10 +3,10 @@
 from datetime import UTC, datetime
 from decimal import Decimal
 from enum import Enum
-from typing import Any
+from typing import Any, Literal
 from uuid import UUID, uuid4
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
 
 
 class OrderType(str, Enum):
@@ -49,8 +49,22 @@ class OrderRequest(BaseModel):
     limit_price: float | None = None
     stop_price: float | None = None
 
+    order_class: Literal["simple", "bracket"] = "simple"
+    take_profit_price: float | None = None
+    stop_loss_price: float | None = None
+
     model_config = {"from_attributes": True}
 
+    @model_validator(mode="after")
+    def _bracket_requires_legs(self) -> "OrderRequest":
+        if self.order_class == "bracket" and (
+            self.take_profit_price is None or self.stop_loss_price is None
+        ):
+            raise ValueError(
+                "bracket orders require take_profit_price + stop_loss_price"
+            )
+        return self
+
 
 class OrderResult(BaseModel):
     """Returned after order submission or status query."""
diff --git a/tests/services/trade_executor/__init__.py b/tests/services/trade_executor/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/services/trade_executor/test_risk_manager_kevin_caps.py b/tests/services/trade_executor/test_risk_manager_kevin_caps.py
new file mode 100644
index 0000000..4d17dc5
--- /dev/null
+++ b/tests/services/trade_executor/test_risk_manager_kevin_caps.py
@@ -0,0 +1,248 @@
+"""Tests for the Kevin-specific risk caps (Task 19).
+
+Daily trade-count cap, daily allocation cap, drawdown halt, daily-loss
+circuit breaker. The Kevin caps are namespaced inside `check_risk` so
+they only apply when the incoming `TradeSignal.strategy_id ==
+KEVIN_STRATEGY_UUID`.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from decimal import Decimal
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from services.trade_executor.config import TradeExecutorConfig
+from services.trade_executor.risk_manager import RiskManager
+from shared.constants.kevin import KEVIN_STRATEGY_UUID
+from shared.schemas.trading import (
+    AccountInfo,
+    PositionInfo,
+    SignalDirection,
+    TradeSignal,
+)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _kevin_config(**overrides) -> TradeExecutorConfig:
+    defaults = dict(
+        max_position_pct=0.05,
+        max_total_exposure_pct=0.80,
+        max_positions=20,
+        default_stop_loss_pct=0.03,
+        cooldown_minutes=30,
+        alpaca_api_key="test",
+        alpaca_secret_key="test",
+        paper_trading=True,
+        kevin_daily_trade_cap=5,
+        kevin_daily_alloc_cap_usd=10000.0,
+        kevin_equity_drawdown_halt_pct=0.20,
+        kevin_daily_loss_circuit_pct=0.05,
+    )
+    defaults.update(overrides)
+    return TradeExecutorConfig(**defaults)
+
+
+def _kevin_signal(target_dollars: float = 2000.0) -> TradeSignal:
+    return TradeSignal(
+        ticker="NVDA",
+        direction=SignalDirection.LONG,
+        strength=0.8,
+        strategy_sources=["meet_kevin"],
+        sentiment_context={"current_price": 150.0},
+        timestamp=datetime.now(timezone.utc),
+        strategy_id=KEVIN_STRATEGY_UUID,
+        target_dollars=Decimal(str(target_dollars)),
+    )
+
+
+def _non_kevin_signal() -> TradeSignal:
+    return TradeSignal(
+        ticker="AAPL",
+        direction=SignalDirection.LONG,
+        strength=0.8,
+        strategy_sources=["news_sentiment"],
+        sentiment_context={"current_price": 150.0},
+        timestamp=datetime.now(timezone.utc),
+        # no strategy_id → not Kevin
+    )
+
+
+def _mock_broker(equity: float = 100_000.0) -> AsyncMock:
+    broker = AsyncMock()
+    broker.get_positions = AsyncMock(return_value=[])
+    broker.get_account = AsyncMock(
+        return_value=AccountInfo(
+            equity=equity, cash=equity, buying_power=equity * 2, portfolio_value=equity
+        )
+    )
+    return broker
+
+
+def _redis_mock(
+    daily_trades: int = 0,
+    daily_alloc: float = 0.0,
+    starting_equity: float | None = 100_000.0,
+    daily_pnl_usd: float = 0.0,
+    paused: bool = False,
+) -> AsyncMock:
+    """Returns an AsyncMock Redis that answers the specific keys the
+    RiskManager queries in the Kevin caps path.
+    """
+    redis = AsyncMock()
+
+    async def _get(key):
+        if isinstance(key, bytes):
+            key = key.decode()
+        # global pause
+        if key == "trading:paused":
+            return b"1" if paused else None
+        # kevin counters (today)
+        if key.startswith("kevin:daily_trades:"):
+            return str(daily_trades).encode()
+        if key.startswith("kevin:daily_alloc_usd:"):
+            return str(daily_alloc).encode()
+        if key == "kevin:starting_equity_usd":
+            return None if starting_equity is None else str(starting_equity).encode()
+        if key.startswith("kevin:daily_pnl_usd:"):
+            return str(daily_pnl_usd).encode()
+        return None
+
+    redis.get = AsyncMock(side_effect=_get)
+    redis.set = AsyncMock()
+    redis.setex = AsyncMock()
+    return redis
+
+
+# ---------------------------------------------------------------------------
+# Tests — daily trade-count cap
+# ---------------------------------------------------------------------------
+
+
+class TestKevinDailyTradeCap:
+    @pytest.mark.asyncio
+    async def test_below_cap_passes(self):
+        config = _kevin_config(kevin_daily_trade_cap=5)
+        broker = _mock_broker()
+        rm = RiskManager(config, broker, redis=_redis_mock(daily_trades=3))
+        with patch.object(RiskManager, "_is_market_hours", return_value=True):
+            approved, _ = await rm.check_risk(_kevin_signal())
+        assert approved is True
+
+    @pytest.mark.asyncio
+    async def test_at_cap_rejects(self):
+        config = _kevin_config(kevin_daily_trade_cap=5)
+        broker = _mock_broker()
+        rm = RiskManager(config, broker, redis=_redis_mock(daily_trades=5))
+        with patch.object(RiskManager, "_is_market_hours", return_value=True):
+            approved, reason = await rm.check_risk(_kevin_signal())
+        assert approved is False
+        assert reason == "kevin_daily_trade_cap"
+
+    @pytest.mark.asyncio
+    async def test_cap_does_not_apply_to_non_kevin_signal(self):
+        config = _kevin_config(kevin_daily_trade_cap=5)
+        broker = _mock_broker()
+        rm = RiskManager(config, broker, redis=_redis_mock(daily_trades=999))
+        with patch.object(RiskManager, "_is_market_hours", return_value=True):
+            approved, _ = await rm.check_risk(_non_kevin_signal())
+        assert approved is True
+
+
+# ---------------------------------------------------------------------------
+# Tests — daily allocation cap
+# ---------------------------------------------------------------------------
+
+
+class TestKevinDailyAllocCap:
+    @pytest.mark.asyncio
+    async def test_within_alloc_cap_passes(self):
+        config = _kevin_config(kevin_daily_alloc_cap_usd=10_000)
+        broker = _mock_broker()
+        rm = RiskManager(config, broker, redis=_redis_mock(daily_alloc=6000))
+        with patch.object(RiskManager, "_is_market_hours", return_value=True):
+            approved, _ = await rm.check_risk(_kevin_signal(target_dollars=2000))
+        assert approved is True
+
+    @pytest.mark.asyncio
+    async def test_alloc_cap_rejects(self):
+        config = _kevin_config(kevin_daily_alloc_cap_usd=10_000)
+        broker = _mock_broker()
+        rm = RiskManager(config, broker, redis=_redis_mock(daily_alloc=9000))
+        with patch.object(RiskManager, "_is_market_hours", return_value=True):
+            # 9000 + 2000 = 11000 > 10000
+            approved, reason = await rm.check_risk(_kevin_signal(target_dollars=2000))
+        assert approved is False
+        assert reason == "kevin_daily_alloc_cap"
+
+
+# ---------------------------------------------------------------------------
+# Tests — drawdown halt (permanent pause)
+# ---------------------------------------------------------------------------
+
+
+class TestKevinDrawdownHalt:
+    @pytest.mark.asyncio
+    async def test_drawdown_below_threshold_sets_permanent_pause(self):
+        config = _kevin_config(kevin_equity_drawdown_halt_pct=0.20)
+        broker = _mock_broker(equity=70_000)  # 30% drawdown from 100k
+        redis = _redis_mock(starting_equity=100_000)
+        rm = RiskManager(config, broker, redis=redis)
+        with patch.object(RiskManager, "_is_market_hours", return_value=True):
+            approved, reason = await rm.check_risk(_kevin_signal())
+        assert approved is False
+        assert reason == "kevin_drawdown_halt"
+        # permanent pause was set (no TTL)
+        redis.set.assert_any_call("trading:paused", "1")
+
+    @pytest.mark.asyncio
+    async def test_drawdown_above_threshold_passes(self):
+        config = _kevin_config(kevin_equity_drawdown_halt_pct=0.20)
+        broker = _mock_broker(equity=85_000)  # 15% drawdown → above threshold
+        rm = RiskManager(
+            config, broker, redis=_redis_mock(starting_equity=100_000)
+        )
+        with patch.object(RiskManager, "_is_market_hours", return_value=True):
+            approved, _ = await rm.check_risk(_kevin_signal())
+        assert approved is True
+
+
+# ---------------------------------------------------------------------------
+# Tests — daily-loss circuit breaker (24h pause)
+# ---------------------------------------------------------------------------
+
+
+class TestKevinDailyLossCircuit:
+    @pytest.mark.asyncio
+    async def test_daily_loss_exceeds_circuit_sets_24h_pause(self):
+        config = _kevin_config(kevin_daily_loss_circuit_pct=0.05)
+        broker = _mock_broker(equity=100_000)
+        # -6% daily loss on 100k equity = -6000
+        redis = _redis_mock(daily_pnl_usd=-6000)
+        rm = RiskManager(config, broker, redis=redis)
+        with patch.object(RiskManager, "_is_market_hours", return_value=True):
+            approved, reason = await rm.check_risk(_kevin_signal())
+        assert approved is False
+        assert reason == "kevin_daily_loss_circuit"
+        # 24h pause set via setex
+        assert redis.setex.called
+        # called with ttl ~= 86400
+        args, _kw = redis.setex.call_args
+        assert args[0] == "trading:paused"
+        assert int(args[1]) == 86400
+
+    @pytest.mark.asyncio
+    async def test_daily_loss_within_circuit_passes(self):
+        config = _kevin_config(kevin_daily_loss_circuit_pct=0.05)
+        broker = _mock_broker(equity=100_000)
+        # -3% daily loss → below circuit
+        rm = RiskManager(config, broker, redis=_redis_mock(daily_pnl_usd=-3000))
+        with patch.object(RiskManager, "_is_market_hours", return_value=True):
+            approved, _ = await rm.check_risk(_kevin_signal())
+        assert approved is True
diff --git a/tests/shared/broker/__init__.py b/tests/shared/broker/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/shared/broker/test_alpaca_bracket.py b/tests/shared/broker/test_alpaca_bracket.py
new file mode 100644
index 0000000..f4a8709
--- /dev/null
+++ b/tests/shared/broker/test_alpaca_bracket.py
@@ -0,0 +1,89 @@
+"""Unit tests for BRACKET order support in OrderRequest + AlpacaBroker.
+
+No network — exercises the request-building path only.
+"""
+
+from __future__ import annotations
+
+import pytest
+from alpaca.trading.enums import OrderClass as AlpacaOrderClass
+from alpaca.trading.requests import MarketOrderRequest
+from pydantic import ValidationError
+
+from shared.broker.alpaca_broker import AlpacaBroker
+from shared.schemas.trading import OrderRequest, OrderSide, OrderType
+
+
+def _broker() -> AlpacaBroker:
+    return AlpacaBroker(api_key="test", secret_key="test", paper=True)
+
+
+def test_order_request_defaults_to_simple_class():
+    o = OrderRequest(ticker="NVDA", side=OrderSide.BUY, qty=10)
+    assert o.order_class == "simple"
+    assert o.take_profit_price is None
+    assert o.stop_loss_price is None
+
+
+def test_order_request_bracket_requires_both_legs():
+    with pytest.raises(ValidationError, match="bracket orders require"):
+        OrderRequest(
+            ticker="NVDA",
+            side=OrderSide.BUY,
+            qty=10,
+            order_class="bracket",
+            take_profit_price=200.0,
+            # stop_loss_price missing
+        )
+
+    with pytest.raises(ValidationError, match="bracket orders require"):
+        OrderRequest(
+            ticker="NVDA",
+            side=OrderSide.BUY,
+            qty=10,
+            order_class="bracket",
+            stop_loss_price=150.0,
+            # take_profit_price missing
+        )
+
+
+def test_order_request_bracket_with_both_legs_validates():
+    o = OrderRequest(
+        ticker="NVDA",
+        side=OrderSide.BUY,
+        qty=10,
+        order_class="bracket",
+        take_profit_price=200.0,
+        stop_loss_price=150.0,
+    )
+    assert o.order_class == "bracket"
+    assert o.take_profit_price == 200.0
+    assert o.stop_loss_price == 150.0
+
+
+def test_build_order_request_simple_market():
+    o = OrderRequest(ticker="NVDA", side=OrderSide.BUY, qty=10)
+    req = _broker()._build_order_request(o)
+    assert isinstance(req, MarketOrderRequest)
+    assert req.symbol == "NVDA"
+    assert req.qty == 10
+    # default MarketOrderRequest has order_class=None or SIMPLE
+    assert getattr(req, "order_class", None) in (None, AlpacaOrderClass.SIMPLE)
+
+
+def test_build_order_request_bracket_attaches_legs():
+    o = OrderRequest(
+        ticker="NVDA",
+        side=OrderSide.BUY,
+        qty=10,
+        order_class="bracket",
+        take_profit_price=200.0,
+        stop_loss_price=150.0,
+    )
+    req = _broker()._build_order_request(o)
+    assert isinstance(req, MarketOrderRequest)
+    assert req.order_class == AlpacaOrderClass.BRACKET
+    assert req.take_profit is not None
+    assert float(req.take_profit.limit_price) == 200.0
+    assert req.stop_loss is not None
+    assert float(req.stop_loss.stop_price) == 150.0