组合持仓退出规则与投资组合风险限额
笔记本 《交易机器学习》
总结
本笔记介绍如何评估持仓层面的退出规则，并将其与投资组合整体控制相结合。固定止损、止盈和时间退出规则与移动止损、收紧止损形成对照；分档退出则在连续达到盈利阈值时逐步减仓。RuleChain 按优先顺序应用第一个触发的规则，而 AllOf 要求所有规则都触发。这些示例展示了策略选择如何影响记录的操作和退出原因。
回撤、单日亏损、总敞口或净敞口等投资组合限额作用于整个组合，而非单笔交易。笔记强调，限额需要基于授权范围设定阈值，并建立通知、覆盖和恢复交易的治理机制。其证据包括构造的持仓状态和单个投资组合快照，而非历史回测。因此，它无法证明这些退出规则是否能提高收益或会产生多少成本；文中也未评估真实路径频率、执行成本和触及限额后的行为。
核心观点
- 如果多个规则在同一根柱线上触发，规则链的顺序决定记录哪种退出。
- 移动止损使用此前已完成柱线的高点，避免依赖下单时尚不可用的信息。
- 分档退出需要保留状态，在用于不同持仓前应重置。
- 持仓退出规则和投资组合限额针对不同风险，应分别评估。
- 总敞口衡量持仓绝对值之和，净敞口衡量带方向符号的敞口。
标签
全文
# Composable Exit Rules and Portfolio Limits
# Composable Exit Rules and Portfolio Limits
**Docker image**: `ml4t`
**Chapter 19: Risk Management**
## Purpose
Use the `ml4t.backtest.risk` library to evaluate position-level exit rules,
compose rule priorities, and evaluate portfolio-level risk limits.
## Learning Objectives
After completing this notebook, you will be able to:
- Configure position-level exit rules (`StopLoss`, `TrailingStop`, `TakeProfit`)
- Compose multiple rules with priority and boolean logic (`RuleChain`, `AllOf`, `AnyOf`)
- Set up portfolio-wide risk limits (`MaxDrawdownLimit`, `DailyLossLimit`)
- Interpret each rule's action, fill price, and reason
## Book reference
Sections 19.4 (drawdowns and recovery), 19.7 (adaptive controls), and 19.8
(kill switches and governance).
## Prerequisites
- Comfort with the backtest framework introduced in Chapter 16
- `02_exit_strategies` and `03_position_sizing_mae_mfe` for the rules'
analytical motivation
```python
"""Configure and evaluate position rules and portfolio limits."""
import numpy as np
import plotly.graph_objects as go
import polars as pl
from ml4t.backtest.risk import (
AllOf,
AnyOf,
DailyLossLimit,
GrossExposureLimit,
MaxDrawdownLimit,
MaxPositionsLimit,
NetExposureLimit,
PortfolioState,
PositionState,
RuleChain,
ScaledExit,
StopLoss,
TakeProfit,
TighteningTrailingStop,
TimeExit,
TrailingStop,
)
from data import load_etfs
from utils.style import COLORS, ml4t_palette, show_plotly_with_alt
```
```python
N_BARS = 252
```
## 1. Position-Level Exit Rules
The `ml4t.backtest.risk.position` module provides exit rules that operate on
individual positions. Each rule evaluates a `PositionState` and returns a
`PositionAction` (HOLD, EXIT_FULL, EXIT_PARTIAL, or ADJUST_STOP).
### Creating Position States
A compact state factory keeps the examples focused on rule behavior. Long and
short returns use the entry notional as the common denominator.
```python
def create_position_state(
symbol: str = "SPY",
side: str = "long",
entry_price: float = 100.0,
current_price: float = 100.0,
bars_held: int = 0,
bar_open: float | None = None,
bar_high: float | None = None,
bar_low: float | None = None,
high_water_mark: float | None = None,
low_water_mark: float | None = None,
) -> PositionState:
"""Create a PositionState for rule evaluation."""
if side not in {"long", "short"}:
raise ValueError("side must be 'long' or 'short'")
direction = 1.0 if side == "long" else -1.0
unrealized_return = direction * (current_price - entry_price) / entry_price
high_mark = max(entry_price, current_price) if high_water_mark is None else high_water_mark
low_mark = min(entry_price, current_price) if low_water_mark is None else low_water_mark
return PositionState(
asset=symbol,
side=side,
entry_price=entry_price,
current_price=current_price,
bar_open=current_price if bar_open is None else bar_open,
bar_high=current_price if bar_high is None else bar_high,
bar_low=current_price if bar_low is None else bar_low,
quantity=100.0,
initial_quantity=100.0,
unrealized_pnl=direction * (current_price - entry_price) * 100.0,
unrealized_return=unrealized_return,
bars_held=bars_held,
high_water_mark=high_mark,
low_water_mark=low_mark,
max_favorable_excursion=max(0, unrealized_return),
max_adverse_excursion=min(0, unrealized_return),
)
```
### Static Exit Rules
Static rules have fixed thresholds that don't change during the position lifetime.
```python
stop_loss = StopLoss(pct=0.05)
take_profit = TakeProfit(pct=0.10)
time_exit = TimeExit(max_bars=20)
```
**`StopLoss`** exits once the loss from the entry price reaches the percentage it was
configured with. It measures against the entry, so the exit level never moves as the
position gains.
```python
scenarios = [
("Entry price", 100.0, 100.0),
("Down 3%", 100.0, 97.0),
("Down 5% (trigger)", 100.0, 95.0),
("Down 7%", 100.0, 93.0),
]
for name, entry, current in scenarios:
state = create_position_state(entry_price=entry, current_price=current)
action = stop_loss.evaluate(state)
status = "TRIGGERED" if action.action.name != "HOLD" else "hold"
print(f" {name}: entry=${entry}, current=${current} -> {status}")
if action.reason:
print(f" Reason: {action.reason}")
```
**`TakeProfit`** is the mirror image: it exits once the gain from entry reaches its target.
```python
scenarios = [
("Entry price", 100.0, 100.0),
("Up 5%", 100.0, 105.0),
("Above 10% (trigger)", 100.0, 110.01),
("Up 15%", 100.0, 115.0),
]
for name, entry, current in scenarios:
state = create_position_state(entry_price=entry, current_price=current)
action = take_profit.evaluate(state)
status = "TRIGGERED" if action.action.name != "HOLD" else "hold"
print(f" {name}: entry=${entry}, current=${current} -> {status}")
```
**`TimeExit(max_bars=20)`** exits after 20 bars regardless of P&L.
```python
for bars in [5, 15, 19, 20, 25]:
state = create_position_state(bars_held=bars)
action = time_exit.evaluate(state)
status = "TRIGGERED" if action.action.name != "HOLD" else "hold"
print(f" Bars held: {bars} -> {status}")
```
### Dynamic Exit Rules
Dynamic rules have thresholds that adapt to position performance.
**`TrailingStop`** measures from the highest price reached since entry rather than from the
entry itself, so the exit level ratchets up with the position and never back down.
The rule uses the high-water mark through the prior completed bar, then updates
that state after the current bar is evaluated.
```python
trailing_stop = TrailingStop(pct=0.05)
states = [
create_position_state(entry_price=100, current_price=100),
create_position_state(entry_price=100, current_price=110),
create_position_state(entry_price=100, current_price=120), # New high
create_position_state(entry_price=100, current_price=116), # Down 3.3%
create_position_state(entry_price=100, current_price=114), # Down 5% (trigger)
]
hwm = 100.0
for state in states:
state.high_water_mark = hwm
action = trailing_stop.evaluate(state)
status = "TRIGGERED" if action.action.name != "HOLD" else "hold"
trail_level = hwm * 0.95
print(f" Price ${state.current_price}: prior HWM=${hwm}, trail=${trail_level:.1f} -> {status}")
hwm = max(hwm, state.current_price)
```
**`TighteningTrailingStop`** narrows the trail as the position gains, on a schedule of
(return reached, trail width) pairs. The reasoning is that an unrealized gain is worth
protecting more tightly than an unproven one, at the cost of being stopped out of a
position that was going to keep running.
```python
tightening = TighteningTrailingStop(
[
(0.00, 0.05),
(0.10, 0.03),
(0.20, 0.02),
]
)
for peak_price in [108.0, 116.0, 128.0]:
current_price = peak_price * 0.96
state = create_position_state(
entry_price=100,
current_price=current_price,
high_water_mark=peak_price,
)
action = tightening.evaluate(state)
status = action.reason if action.reason else "hold"
print(f" Peak ${peak_price:.0f}, current ${current_price:.2f}: {status}")
```
**`ScaledExit`** sells part of the position at each of a series of profit targets, taking a
stated fraction of what remains at each one. Unlike every other rule here it is stateful:
it remembers which targets have already fired, so one instance serves a single position and
must be reset before it is used for another.
```python
scaled = ScaledExit(
[
(0.05, 0.25),
(0.10, 0.33),
(0.15, 0.50),
]
)
# Use prices just beyond each decimal threshold so binary floating-point
# representation cannot turn an intended crossing into an equality artifact.
for return_pct in [0.03, 0.0501, 0.08, 0.1001, 0.1501]:
state = create_position_state(entry_price=100, current_price=100 * (1 + return_pct))
action = scaled.evaluate(state)
if action.action.name == "EXIT_PARTIAL":
print(f" At {return_pct:.2%}: EXIT {action.pct:.0%} of position ({action.reason})")
else:
print(f" At {return_pct:.2%}: hold")
scaled.reset()
```
## 2. Rule Composition
The `ml4t.backtest.risk` module provides composition patterns to combine rules:
- **RuleChain**: First non-HOLD wins (priority order)
- **AllOf**: All must trigger (AND logic)
- **AnyOf**: Any can trigger (OR logic, alias for RuleChain)
**`RuleChain`** evaluates its rules in order and returns the first one that does not say
hold. Order is therefore policy: putting the stop first means a bar breaching both the stop
and the target is recorded as a stop, which is the conservative reading of a bar whose
internal sequence is unknown.
```python
chain = RuleChain(
[
StopLoss(pct=0.05),
TakeProfit(pct=0.10),
TrailingStop(pct=0.03),
TimeExit(max_bars=20),
]
)
# Test scenarios
test_cases = [
("Loss triggers stop", 100, 94, 5, 100), # Stop loss
("Profit triggers TP", 100, 112, 5, 112), # Take profit
("Trail triggers", 100, 108, 5, 115), # Fell from 115 to 108 (>3%)
("Time triggers", 100, 102, 22, 105), # Held 22 bars
("Nothing triggers", 100, 102, 5, 102), # All hold
]
for name, entry, current, bars, hwm in test_cases:
state = create_position_state(entry_price=entry, current_price=current, bars_held=bars)
state.high_water_mark = hwm
action = chain.evaluate(state)
if action.action.name != "HOLD":
print(f" {name}: {action.reason}")
else:
print(f" {name}: HOLD")
```
**`AllOf`** requires every rule to trigger before it exits. The example below combines a
gain threshold with a minimum holding period, so neither a quick gain nor a long flat hold
exits on its own.
```python
all_of = AllOf(
[
TakeProfit(pct=0.01),
TimeExit(max_bars=5),
]
)
test_cases = [
("Profitable, 3 bars", 0.05, 3),
("Profitable, 5 bars", 0.05, 5),
("Loss, 10 bars", -0.02, 10),
("Breakeven, 5 bars", 0.0, 5),
]
for name, ret, bars in test_cases:
state = create_position_state(entry_price=100, current_price=100 * (1 + ret), bars_held=bars)
state.unrealized_return = ret
action = all_of.evaluate(state)
status = "EXIT" if action.action.name != "HOLD" else "HOLD"
print(f" {name} (ret={ret:.0%}, bars={bars}): {status}")
```
**`AnyOf`** exits when any rule triggers, which makes it equivalent to `RuleChain`. Worth
carrying over from `02_exit_strategies`: a rule that fires on almost every bar dominates
any combination it is placed in.
```python
any_of = AnyOf(
[
StopLoss(pct=0.05),
TakeProfit(pct=0.10),
TimeExit(max_bars=20),
]
)
{
"AnyOf type": type(any_of).__name__,
"rules in chain": len(any_of.rules),
"doc": (AnyOf.__doc__ or "").strip().splitlines()[0] if AnyOf.__doc__ else "",
}
```
## 3. Portfolio-Level Limits
Portfolio limits operate on the entire portfolio state, not individual positions.
They implement **kill switches** and **guardrails** discussed in Section 19.8.
`create_portfolio_state` packages the inputs (equity, high-water mark,
positions, daily P&L) into a `PortfolioState` so each limit check has a
single object to reason about. We use it as a scaffold in the demos
below; in production the backtester reconstructs `PortfolioState` from
the broker on every bar.
```python
def create_portfolio_state(
equity: float = 100000,
initial_equity: float = 100000,
high_water_mark: float = 100000,
num_positions: int = 5,
positions: dict[str, float] | None = None,
daily_pnl: float = 0,
) -> PortfolioState:
"""Create a PortfolioState for limit checks."""
if positions is None:
positions = {f"ASSET_{i}": equity / 10 for i in range(num_positions)}
else:
num_positions = len(positions)
gross = sum(abs(v) for v in positions.values())
net = sum(positions.values())
drawdown = (
(high_water_mark - equity) / high_water_mark
if high_water_mark > 0 and equity < high_water_mark
else 0.0
)
return PortfolioState(
equity=equity,
initial_equity=initial_equity,
high_water_mark=high_water_mark,
current_drawdown=drawdown,
num_positions=num_positions,
positions=positions,
daily_pnl=daily_pnl,
gross_exposure=gross,
net_exposure=net,
)
```
**`MaxDrawdownLimit`** carries two thresholds, a warning level and a liquidation level, so
the escalation from "look at this" to "close everything" is written into the rule rather
than decided under pressure.
```python
dd_limit = MaxDrawdownLimit(max_drawdown=0.20, warn_threshold=0.15)
for dd_pct in [0.05, 0.10, 0.15, 0.18, 0.20, 0.25]:
equity = 100000 * (1 - dd_pct)
state = create_portfolio_state(equity=equity, high_water_mark=100000)
result = dd_limit.check(state)
if result.breached:
print(f" Drawdown {dd_pct:.0%}: {result.action.upper()} - {result.reason}")
else:
print(f" Drawdown {dd_pct:.0%}: OK")
```
**`DailyLossLimit`** liquidates once the session's loss exceeds its threshold as a share of
current equity. Measuring against current rather than starting equity means the limit
tightens in absolute terms as the book shrinks.
```python
daily_limit = DailyLossLimit(max_daily_loss_pct=0.02)
for daily_pnl in [500, 0, -1000, -2000, -2500]:
state = create_portfolio_state(equity=100000, daily_pnl=daily_pnl)
result = daily_limit.check(state)
pct = daily_pnl / 100000 * 100
if result.breached:
print(f" Daily P&L ${daily_pnl:+,} ({pct:+.1f}%): {result.action.upper()}")
else:
print(f" Daily P&L ${daily_pnl:+,} ({pct:+.1f}%): OK")
```
**`MaxPositionsLimit`** halts when the open-position count reaches 10.
```python
pos_limit = MaxPositionsLimit(max_positions=10)
for n_pos in [5, 8, 10, 12]:
positions = {f"ASSET_{i}": 10000 for i in range(n_pos)}
state = create_portfolio_state(num_positions=n_pos, positions=positions)
result = pos_limit.check(state)
if result.breached:
print(f" {n_pos} positions: {result.action.upper()} - {result.reason}")
else:
print(f" {n_pos} positions: OK")
```
**`GrossExposureLimit`** halts new trading above its threshold. Gross exposure adds the
absolute value of every position, so a long and a short of equal size count double rather
than netting to nothing - which is the point, since both can lose at once.
```python
gross_limit = GrossExposureLimit(max_gross_exposure=1.5)
for leverage in [0.8, 1.0, 1.3, 1.5, 2.0]:
positions = {"LONG": 100000 * leverage / 2, "SHORT": -100000 * leverage / 2}
state = create_portfolio_state(
equity=100000,
positions=positions,
num_positions=2,
)
result = gross_limit.check(state)
if result.breached:
print(f" {leverage:.0%} gross: {result.action.upper()} - {result.reason}")
else:
print(f" {leverage:.0%} gross: OK")
```
**`NetExposureLimit`** bounds the signed sum instead, which is directional market exposure.
A book can sit inside a tight net band while carrying large gross positions, so the two
limits constrain different things and a portfolio needs both.
```python
net_limit = NetExposureLimit(max_net_exposure=0.10, min_net_exposure=-0.10)
test_cases = [
("Neutral", {"LONG": 50000, "SHORT": -50000}),
("+5% net", {"LONG": 55000, "SHORT": -50000}),
("+15% net", {"LONG": 60000, "SHORT": -45000}),
("-12% net", {"LONG": 44000, "SHORT": -56000}),
]
for name, positions in test_cases:
state = create_portfolio_state(equity=100000, positions=positions, num_positions=2)
result = net_limit.check(state)
if result.breached:
print(f" {name}: {result.action.upper()} - {result.reason}")
else:
print(f" {name}: OK")
```
## 4. Practical Example: Layered Rule Configuration
Position rules and portfolio limits address different decisions. The configuration
below illustrates both layers; governance still requires escalation, approval, and
reinstatement procedures outside these classes.
```python
# Define position-level rules
position_rules = RuleChain(
[
StopLoss(pct=0.03), # 3% hard stop
TighteningTrailingStop(
[
(0.00, 0.05), # 5% trail initially
(0.10, 0.03), # Tighten to 3% at +10%
(0.20, 0.02), # Tighten to 2% at +20%
]
),
TakeProfit(pct=0.30), # 30% take profit
TimeExit(max_bars=60), # Exit after 60 bars
]
)
# Define portfolio-level limits
portfolio_limits = [
MaxDrawdownLimit(max_drawdown=0.15, warn_threshold=0.10),
DailyLossLimit(max_daily_loss_pct=0.02),
MaxPositionsLimit(max_positions=20),
GrossExposureLimit(max_gross_exposure=1.0),
]
```
The two layers answer different questions and both are needed. The position rules decide what to
do about one trade going wrong, in priority order: a hard stop measured from entry, a trail that
tightens as the gain grows, a final target, and a holding-period cap so nothing is carried
indefinitely. The portfolio limits decide when to stop trading altogether regardless of which
individual position is responsible - a drawdown from the equity high, a single session's loss,
a count of open positions, and total gross exposure.
A position rule cannot see the portfolio and a portfolio limit cannot see which trade to close.
The configured values are in the cells above; the cell below prints them back so the
configuration a reader is looking at is the one being evaluated.
```python
print("Position rules, in priority order:")
for rule in position_rules.rules:
print(f" {type(rule).__name__}")
print("\nPortfolio limits:")
for limit in portfolio_limits:
print(f" {type(limit).__name__}")
```
### Evaluate Illustrative Position Paths
```python
positions_sim = [
{
"symbol": "AAPL",
"entry_price": 150.0,
"prices": [150, 148, 146, 145.5, 145], # Declining -> stop loss
},
{
"symbol": "GOOGL",
"entry_price": 100.0,
"prices": [100, 108, 115, 118, 113], # Rise then trail triggers
},
{
"symbol": "MSFT",
"entry_price": 300.0,
"prices": [300, 305, 310, 308, 312], # Steady rise, hold
},
]
```
```python
for pos in positions_sim:
print(f"\n{pos['symbol']} (entry: ${pos['entry_price']})")
hwm = pos["entry_price"]
for i, price in enumerate(pos["prices"]):
state = create_position_state(
symbol=pos["symbol"],
entry_price=pos["entry_price"],
current_price=price,
bars_held=i,
high_water_mark=hwm,
)
action = position_rules.evaluate(state)
ret = (price / pos["entry_price"] - 1) * 100
if action.action.name != "HOLD":
print(f" Bar {i}: ${price} ({ret:+.1f}%) HWM=${hwm:.0f} -> EXIT: {action.reason}")
break
else:
print(f" Bar {i}: ${price} ({ret:+.1f}%) HWM=${hwm:.0f} -> hold")
hwm = max(hwm, price)
```
### Portfolio Limit Check
```python
portfolio = create_portfolio_state(
equity=92000, # Down from 100k
initial_equity=100000,
high_water_mark=105000, # Was up 5% at peak
daily_pnl=-1800, # Down $1,800 today
num_positions=8,
positions={f"POS_{i}": 11500 for i in range(8)}, # 100% gross exposure
)
print("\nPortfolio State:")
print(f" Equity: ${portfolio.equity:,.0f}")
print(f" High Water Mark: ${portfolio.high_water_mark:,.0f}")
print(f" Current Drawdown: {portfolio.current_drawdown:.1%}")
print(f" Daily P&L: ${portfolio.daily_pnl:+,.0f} ({portfolio.daily_pnl / portfolio.equity:.1%})")
print(f" Positions: {portfolio.num_positions}")
print(f" Gross Exposure: {portfolio.gross_exposure / portfolio.equity:.0%}")
print("\nLimit Checks:")
for limit in portfolio_limits:
result = limit.check(portfolio)
limit_name = limit.__class__.__name__
if result.breached:
print(f" {limit_name}: {result.action.upper()} - {result.reason}")
else:
print(f" {limit_name}: OK")
```
## 5. Rule Diagnostics
The figures below visualize rule mechanics, not strategy performance. Thresholds
are fixed before evaluation, and every surface calls the library on controlled
states rather than clipping or rewriting realized returns.
### Trigger Timing on a Real Price Path
The position is entered after the first SPY close of 2020. Each subsequent bar
supplies its observed OHLC range. The trailing rule receives the high-water mark
through the previous completed bar, which preserves the library's default lagged
timing and avoids using the current bar's high before evaluating its low.
```python
spy = (
load_etfs(symbols=["SPY"], start_date="2020-01-01", end_date="2020-12-31")
.sort("timestamp")
.head(N_BARS)
)
entry_date = spy.item(0, "timestamp")
entry_price = float(spy.item(0, "close"))
print(f"Entry after {entry_date}: SPY close ${entry_price:.2f}; bars loaded: {spy.height}")
```
`first_trigger` makes the event order explicit. The entry bar is never tested,
current-bar OHLC is observable to active stop orders, and water marks advance only
after a bar completes without an exit.
```python
def first_trigger(frame: pl.DataFrame, rule_name: str, rule: object) -> dict[str, object]:
"""Return the first action from a rule on a long position entered at the first close."""
entry = float(frame.item(0, "close"))
high_water_mark = entry
low_water_mark = entry
for bars_held, bar in enumerate(frame.iter_rows(named=True), start=0):
if bars_held == 0:
continue
state = create_position_state(
entry_price=entry,
current_price=float(bar["close"]),
bars_held=bars_held,
bar_open=float(bar["open"]),
bar_high=float(bar["high"]),
bar_low=float(bar["low"]),
high_water_mark=high_water_mark,
low_water_mark=low_water_mark,
)
action = rule.evaluate(state)
if action.action.name != "HOLD":
return {
"rule": rule_name,
"timestamp": bar["timestamp"],
"bars_held": bars_held,
"close": float(bar["close"]),
"fill_price": action.fill_price,
"reason": action.reason,
}
high_water_mark = max(high_water_mark, float(bar["high"]))
low_water_mark = min(low_water_mark, float(bar["low"]))
raise RuntimeError(f"{rule_name} did not trigger within the observed path")
```
The same entry and bars feed three independent rules. Their first actions retain
the library's fill-price convention and human-readable reason.
```python
timeline_rules = {
"Stop loss 5%": StopLoss(pct=0.05),
"Trailing stop 3%": TrailingStop(pct=0.03),
"Take profit 15%": TakeProfit(pct=0.15),
}
trigger_results = pl.DataFrame(
[first_trigger(spy, rule_name, rule) for rule_name, rule in timeline_rules.items()]
).sort("timestamp")
trigger_results
```
The trigger markers use fill prices, while the line shows daily closes. A shared
date axis prevents nearby selloff triggers from collapsing into overlapping text.
```python
fig = go.Figure()
fig.add_trace(
go.Scatter(
x=spy["timestamp"].to_list(),
y=spy["close"].to_list(),
name="SPY close",
line=dict(color=COLORS["neutral"], width=2),
)
)
for result, color in zip(
trigger_results.iter_rows(named=True),
ml4t_palette(trigger_results.height, categorical=True),
strict=True,
):
fig.add_trace(
go.Scatter(
x=[result["timestamp"]],
y=[result["fill_price"]],
mode="markers",
name=result["rule"],
marker=dict(color=color, size=11, line=dict(color=COLORS["bg_light"], width=1)),
hovertemplate=(
f"{result['rule']}<br>%{{x|%Y-%m-%d}}<br>Fill $%{{y:.2f}}<extra></extra>"
),
)
)
fig.update_layout(
title=(
"SPY Trigger Dates Differ Across Exit Rules"
"<br><sup>Long entry after the first 2020 close; default stop-price fills</sup>"
),
xaxis_title="Date",
yaxis_title="SPY price ($)",
height=430,
legend_title_text="First action",
)
show_plotly_with_alt(
fig,
"A price path with markers showing where each rule in the chain would have triggered, so the order in which they fire is visible against the same series.",
)
```
### Rule Priority Across Position States
A close-only grid isolates composition semantics. When the 20-bar time exit
overlaps a price rule, `RuleChain` returns the earlier rule in its declared order.
```python
return_grid = np.arange(-10, 16, 1)
bars_grid = np.arange(0, 31, 2)
exit_chain = RuleChain([StopLoss(pct=0.05), TakeProfit(pct=0.10), TimeExit(max_bars=20)])
exit_code = {"HOLD": 0, "stop_loss": 1, "take_profit": 2, "time_exit": 3}
exit_surface = np.zeros((len(bars_grid), len(return_grid)), dtype=int)
for row_idx, bars_held in enumerate(bars_grid):
for col_idx, return_pct in enumerate(return_grid):
state = create_position_state(
entry_price=100.0,
current_price=100.0 * (1 + return_pct / 100),
bars_held=int(bars_held),
)
action = exit_chain.evaluate(state)
reason_key = next((key for key in exit_code if action.reason.startswith(key)), "HOLD")
exit_surface[row_idx, col_idx] = exit_code[reason_key]
```
The categorical map separates `HOLD`, entry stop, profit target, and time exit.
Its colors encode actions rather than performance.
```python
exit_colors = [
COLORS["silver_muted"],
COLORS["negative"],
COLORS["positive"],
COLORS["amber"],
]
exit_scale = [
[0.0, exit_colors[0]],
[1 / 6, exit_colors[0]],
[1 / 6, exit_colors[1]],
[0.5, exit_colors[1]],
[0.5, exit_colors[2]],
[5 / 6, exit_colors[2]],
[5 / 6, exit_colors[3]],
[1.0, exit_colors[3]],
]
```
The colorbar names each action directly so the categorical codes never require
interpretation from the reader.
```python
fig = go.Figure(
go.Heatmap(
x=return_grid,
y=bars_grid,
z=exit_surface,
zmin=0,
zmax=3,
colorscale=exit_scale,
colorbar=dict(
title="First action",
tickmode="array",
tickvals=[0, 1, 2, 3],
ticktext=["HOLD", "STOP", "PROFIT", "TIME"],
),
hovertemplate="Return %{x}%<br>Bars %{y}<br>Action code %{z}<extra></extra>",
)
)
fig.update_layout(
title=(
"Rule Priority Resolves Overlapping Exit Conditions"
"<br><sup>Close-only state grid; stop loss precedes take profit and time exit</sup>"
),
xaxis_title="Return from entry (%)",
yaxis_title="Bars held",
height=430,
)
show_plotly_with_alt(
fig,
"A grid of position states by rule, coloured by the action each rule returns, showing which rule claims each region of the state space under the chain's priority order.",
)
```
### Portfolio Escalation Surface
The portfolio map combines two independent checks without inventing realized
returns. Each cell constructs a `PortfolioState`, calls both limits, and reports
the more severe action if both are breached.
```python
drawdown_grid = np.arange(0, 26, 2.5)
daily_loss_grid = np.arange(0, 4.1, 0.5)
portfolio_surface = np.zeros((len(drawdown_grid), len(daily_loss_grid)), dtype=int)
for row_idx, drawdown_pct in enumerate(drawdown_grid):
for col_idx, daily_loss_pct in enumerate(daily_loss_grid):
equity = 100000 * (1 - drawdown_pct / 100)
state = create_portfolio_state(
equity=equity,
high_water_mark=100000,
daily_pnl=-equity * daily_loss_pct / 100,
num_positions=0,
positions={},
)
actions = {
result.action
for result in (dd_limit.check(state), daily_limit.check(state))
if result.breached
}
portfolio_surface[row_idx, col_idx] = (
2 if "liquidate" in actions else 1 if "warn" in actions else 0
)
```
The boundaries in the surface are the class contracts, not approximations of them: each limit
warns at one configured level and liquidates at another, and the daily-loss limit escalates
straight to liquidation with no warning band at all.
```python
portfolio_colors = [COLORS["silver_muted"], COLORS["amber"], COLORS["negative"]]
portfolio_scale = [
[0.0, portfolio_colors[0]],
[0.25, portfolio_colors[0]],
[0.25, portfolio_colors[1]],
[0.75, portfolio_colors[1]],
[0.75, portfolio_colors[2]],
[1.0, portfolio_colors[2]],
]
fig = go.Figure(
go.Heatmap(
x=daily_loss_grid,
y=drawdown_grid,
z=portfolio_surface,
zmin=0,
zmax=2,
colorscale=portfolio_scale,
colorbar=dict(
title="Action",
tickmode="array",
tickvals=[0, 1, 2],
ticktext=["OK", "WARN", "LIQUIDATE"],
),
hovertemplate="Daily loss %{x}%<br>Drawdown %{y}%<extra></extra>",
)
)
fig.update_layout(
title=(
"Drawdown and Daily-Loss Limits Form an Escalation Surface"
"<br><sup>MaxDrawdownLimit(20%, warn 15%) and DailyLossLimit(2%)</sup>"
),
xaxis_title="Daily loss (% of current equity)",
yaxis_title="Drawdown from high-water mark (%)",
height=430,
)
show_plotly_with_alt(
fig,
"A surface of portfolio state by limit outcome in three bands - no action, warn, and liquidate - with the warning band visible only for the limits that define one.",
)
```
## 6. Demonstrated API Coverage
This inventory summarizes only the classes exercised above. The counts derive
from the names so they cannot drift from the displayed lists.
```python
demonstrated_api = {
"Position rules": (
"StopLoss",
"TakeProfit",
"TimeExit",
"TrailingStop",
"TighteningTrailingStop",
"ScaledExit",
),
"Composition patterns": ("RuleChain", "AllOf", "AnyOf"),
"Portfolio limits": (
"MaxDrawdownLimit",
"DailyLossLimit",
"MaxPositionsLimit",
"GrossExposureLimit",
"NetExposureLimit",
),
}
library_coverage = pl.DataFrame(
{
"category": list(demonstrated_api),
"count": [len(names) for names in demonstrated_api.values()],
"examples": [", ".join(names) for names in demonstrated_api.values()],
}
)
library_coverage
```
**Section integration**:
- §19.4 references `StopLoss`, `TrailingStop`, and `TighteningTrailingStop`
for drawdown and recovery management.
- §19.7 references `RuleChain` for adaptive controls.
- §19.8 references `MaxDrawdownLimit`, `DailyLossLimit` for kill switches.
## Key Takeaways
1. **Exit rules and portfolio limits are different controls and neither substitutes for the
other.** A position rule sees one trade and decides whether to close it. A portfolio limit
sees the book and decides whether to keep trading at all. A book can be full of individually
healthy positions and still be over its drawdown limit, and a single position can need closing
while the book is fine.
2. **Order inside a rule chain is policy, not implementation.** The first rule that fires wins,
so a chain that evaluates the target before the stop will record a bar that breached both as a
profitable exit. Decide the order deliberately and state it, as `02_exit_strategies` had to
for the same reason.
3. **Know which rules carry state.** Most of these evaluate a position afresh each bar. The
scaled exit does not: it remembers which targets have already fired, so reusing one instance
across positions silently skips targets for the second position onward.
4. **Check what each rule is allowed to see on the current bar.** A trailing stop that reads the
current bar's own high to set the water mark it then tests against is using information the
order did not have. The default here takes the water mark from completed bars.
5. **Distinguish gross from net exposure and limit both.** Gross sums absolute positions and net
sums signed ones. A market-neutral book sits near zero net while carrying substantial gross,
and only one of the two limits would catch it levering up.
6. **A limit class enforces a threshold; it does not decide one.** The value comes from a risk
mandate, and the surrounding procedure - who is told, who may override, what has to be true to
resume - is not in the library and is what makes a kill switch usable.
### Known limitations
- Every example here is an evaluation of a constructed state, not a backtest. Nothing measures
what applying these rules would have earned or cost, and the diagnostics show where rules fire
rather than whether firing there was right.
- The position states are built by hand to sit either side of each threshold. That is the right
way to demonstrate a contract and says nothing about how often real paths reach those states.
- No cost is charged on any exit. The rules differ mainly in how often they close positions,
which is exactly what costs price.
- The portfolio limits are checked against a single snapshot of a book rather than a sequence, so
nothing here exercises what happens after a limit breaches and trading halts.
API reference: `ml4t.backtest.risk`.
**Next**: [`11_systematic_risk_sweep`](11_systematic_risk_sweep.ipynb) applies these position
rules across grids of thresholds and reads the resulting surfaces.
在遵守原作品许可的前提下，附作者信息全文展示。 许可协议: MIT
此摘要由 Stratmill 研究智能体根据原文撰写，并非原文副本。