fez5587/momentum-system
1
0 Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

Managed exits (swept + OOS-verified) + bigger armed set

Browse Source 
Replace the static -1R/+2R bracket with active, parameterised exit management
shared by the backtest and live (strategy/exits.py): move to breakeven, trail
under prior-bar lows or by %, scale out, or first-red exit.

Swept all variants over 187 sessions with a held-out train/test split
(sweep_exits.py), then an adversarial 3-judge workflow picked the robust winner
OUT-OF-SAMPLE: trailLow_a1 (trail under the prior bar's low after +1R) — test
avgR 0.128 vs the old static_2R 0.067 (~1.9x), win 49% vs 38%. The naive
in-sample winner (trailPct10) was correctly rejected as overfit
(train 0.684 -> test 0.085).

- strategy/exits.py: ExitConfig + simulate_exit (backtest) + manage_live (live)
- strategy/backtest/engine.py: simulate the managed exit on entry; Trade.entry_index
- runtime/exit_manager.py: live manager — ratchets the broker stop UP via
  REPLACE/PATCH (never cancel = never naked), scales out, first-red exit
- alpaca_paper/client.py: replace_order (PATCH) + partial close_position
- run_live_paper.py: 'exits' step (12s); TRIGGER_MAX_ARMED 6 -> 10
- tests: exits sim/live + manager ratchet-never-cancel

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
momentum devandClaude Opus 4.8 committed 2026-06-17 13:16:47 -04:00
1 parent d5d79cccf7
commit ab5a92447f
8 files changed
+890 -102
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

strategy/backtest/engine.py
+100 -98
View File
@@ -13,6 +13,7 @@ import pandas as pd
from config import BacktestConfig, Config
from strategy.evaluation.setup_evaluator import evaluate_setup
from strategy.exits import ExitConfig, simulate_exit
from strategy.risk.position_sizing import PositionSizingConfig, calculate_position_size
@@ -31,6 +32,7 @@ class Trade:
realized_pnl: float | None = None
r_multiple: float | None = None
setup_type: str | None = None
entry_index: int | None = None # bar position of the entry-fill bar
@dataclass
@@ -68,6 +70,7 @@ class BacktestEngine:
ready_score_pct: float = 60.0,
criteria=None,
min_bars: int = 10,
exit_config: ExitConfig | None = None,
):
self.config = config or Config()
self.bt: BacktestConfig = self.config.backtest
@@ -78,6 +81,8 @@ class BacktestEngine:
self.ready_score_pct = ready_score_pct
self.criteria = criteria
self.min_bars = min_bars
# default reproduces the original static bracket (stop / +target_r / EOD)
self.exit_config = exit_config or ExitConfig(target_r=target_r)
def _entry_costs(self, price: float, quantity: int) -> tuple[float, float]:
slip = price * self.bt.slippage.base_spread_pct
@@ -95,114 +100,111 @@ class BacktestEngine:
) -> BacktestResult:
result = BacktestResult(symbol=symbol)
bars = bars.reset_index(drop=True)
open_trade: Trade | None = None
cfg = self.exit_config
n = len(bars)
i = self.warmup_bars
while i < len(bars) - 1:
bar = bars.iloc[i]
if open_trade is not None:
low, high = float(bar["low"]), float(bar["high"])
exit_price = None
reason = None
if low <= open_trade.stop_price:
exit_price, reason = open_trade.stop_price, "stop_loss"
elif high >= open_trade.target_price:
exit_price, reason = open_trade.target_price, "target"
elif i == len(bars) - 2:
exit_price, reason = float(bar["close"]), "session_end"
if exit_price is not None:
self._close(open_trade, bar, exit_price, reason, result)
open_trade = None
while i < n - 1:
if i % self.eval_every != 0:
i += 1
continue
if i % self.eval_every == 0:
window = bars.iloc[: i + 1]
# Evaluate as of the current bar's timestamp, not wall-clock
# now() — otherwise every setup is judged against the real
# clock and marked "late" past the entry cutoff (so a Sunday
# backtest would show zero trades). Bars are stored UTC-naive;
# the cutoff reasons in US/Eastern, so convert.
eval_time = None
try:
from datetime import timezone as _tz
from zoneinfo import ZoneInfo as _ZI
window = bars.iloc[: i + 1]
# Evaluate as of the current bar's timestamp, not wall-clock now() —
# otherwise every setup is judged against the real clock and marked
# "late" past the entry cutoff (so a Sunday backtest would show zero
# trades). Bars are stored UTC-naive; the cutoff reasons in
# US/Eastern, so convert.
eval_time = None
try:
from datetime import timezone as _tz
from zoneinfo import ZoneInfo as _ZI
last_ts = pd.Timestamp(window["timestamp"].iloc[-1]).to_pydatetime()
if last_ts.tzinfo is None:
last_ts = last_ts.replace(tzinfo=_tz.utc)
eval_time = last_ts.astimezone(_ZI("America/New_York")).replace(
tzinfo=None
)
except Exception: # noqa: BLE001 - fall back to now()
eval_time = None
evaluation = evaluate_setup(
window,
previous_close=previous_close,
avg_daily_volume=avg_daily_volume,
evaluation_time=eval_time,
ready_score_pct=self.ready_score_pct,
criteria=self.criteria,
min_bars=self.min_bars,
last_ts = pd.Timestamp(window["timestamp"].iloc[-1]).to_pydatetime()
if last_ts.tzinfo is None:
last_ts = last_ts.replace(tzinfo=_tz.utc)
eval_time = last_ts.astimezone(_ZI("America/New_York")).replace(
tzinfo=None
)
result.evaluations += 1
if evaluation.status == "ready" and evaluation.setups:
result.signals += 1
setup = evaluation.setups[0]
next_bar = bars.iloc[i + 1]
raw_entry = (
float(next_bar["open"])
if self.bt.entry_mode == "next_bar"
else float(bar["close"])
)
stop = float(setup["stop_loss_price"])
sizing = calculate_position_size(
raw_entry, stop, equity=self.equity,
config=PositionSizingConfig(),
)
if sizing.position_size > 0:
slip, commission = self._entry_costs(
raw_entry, sizing.position_size
)
entry = raw_entry + slip
risk = entry - stop
open_trade = Trade(
trade_id=str(uuid.uuid4()),
symbol=symbol,
entry_time=str(next_bar.get("timestamp", i + 1)),
entry_price=round(entry, 4),
stop_price=round(stop, 4),
target_price=round(entry + self.target_r * risk, 4),
quantity=sizing.position_size,
setup_type=setup.get("setup_type"),
)
open_trade.realized_pnl = -commission
i += 1 # consumed next bar for entry
i += 1
except Exception: # noqa: BLE001 - fall back to now()
eval_time = None
if open_trade is not None:
last = bars.iloc[-1]
self._close(open_trade, last, float(last["close"]), "session_end", result)
evaluation = evaluate_setup(
window,
previous_close=previous_close,
avg_daily_volume=avg_daily_volume,
evaluation_time=eval_time,
ready_score_pct=self.ready_score_pct,
criteria=self.criteria,
min_bars=self.min_bars,
)
result.evaluations += 1
if not (evaluation.status == "ready" and evaluation.setups):
i += 1
continue
result.signals += 1
setup = evaluation.setups[0]
entry_idx = i + 1
next_bar = bars.iloc[entry_idx]
raw_entry = (
float(next_bar["open"])
if self.bt.entry_mode == "next_bar"
else float(bars.iloc[i]["close"])
)
stop = float(setup["stop_loss_price"])
sizing = calculate_position_size(
raw_entry, stop, equity=self.equity, config=PositionSizingConfig(),
)
if sizing.position_size <= 0:
i += 1
continue
slip, entry_commission = self._entry_costs(raw_entry, sizing.position_size)
entry = raw_entry + slip
risk = entry - stop
if risk <= 0:
i += 1
continue
# Simulate the FULL managed exit forward with the shared rules (the
# same logic the live exit manager applies), then resume scanning
# after the exit bar so trades never overlap.
bars_after = bars.iloc[entry_idx + 1:]
res = simulate_exit(entry, stop, bars_after, cfg)
qty = sizing.position_size
gross = res.r_multiple * risk * qty
total_commission = entry_commission
exit_slip = 0.0
for fll in res.fills:
fqty = max(1, int(round(fll.frac * qty)))
total_commission += max(
self.bt.commission.minimum, fqty * self.bt.commission.per_share
)
exit_slip += fll.price * self.bt.slippage.base_spread_pct * fqty
realized = gross - total_commission - exit_slip
exit_pos = min(entry_idx + 1 + res.exit_index, n - 1) if len(bars_after) else entry_idx
trade = Trade(
trade_id=str(uuid.uuid4()),
symbol=symbol,
entry_time=str(next_bar.get("timestamp", entry_idx)),
entry_price=round(entry, 4),
stop_price=round(stop, 4),
target_price=round(entry + cfg.target_r * risk, 4) if cfg.target_r else round(entry, 4),
quantity=qty,
exit_time=str(bars.iloc[exit_pos].get("timestamp", exit_pos)),
exit_price=round(res.fills[-1].price, 4) if res.fills else None,
exit_reason=res.reason,
realized_pnl=round(realized, 2),
r_multiple=round(res.r_multiple, 3),
setup_type=setup.get("setup_type"),
entry_index=entry_idx,
)
result.trades.append(trade)
result.total_pnl += trade.realized_pnl
i = exit_pos + 1 # resume after the exit
if result.trades:
wins = sum(1 for t in result.trades if (t.realized_pnl or 0) > 0)
result.win_rate = wins / len(result.trades)
return result
def _close(
self, trade: Trade, bar, exit_price: float, reason: str | None,
result: BacktestResult,
) -> None:
slip, commission = self._entry_costs(exit_price, trade.quantity)
fill = exit_price - slip
gross = (fill - trade.entry_price) * trade.quantity
trade.exit_time = str(bar.get("timestamp", "end"))
trade.exit_price = round(fill, 4)
trade.exit_reason = reason
trade.realized_pnl = round((trade.realized_pnl or 0.0) + gross - commission, 2)
risk = trade.entry_price - trade.stop_price
trade.r_multiple = round((fill - trade.entry_price) / risk, 3) if risk > 0 else 0.0
result.trades.append(trade)
result.total_pnl += trade.realized_pnl
strategy/exits.py
+235
View File
@@ -0,0 +1,235 @@
"""Trade-exit management — ONE source of truth for the backtest and live.
The original exit was a static bracket: a fixed stop at the opening-range low
(-1R) and a fixed take-profit at +target_r. Once filled, nothing moved, so a
trade could run to +1.8R and round-trip to a full loss, and losers were always
taken at the full stop.
This module adds active management as pure, parameterised logic so the exact
same rules drive the backtest (where we sweep them) and the live manager (where
we apply the winner). Rules, each independently toggleable:
* breakeven — once price reaches +breakeven_at_r, move the stop to entry
* trail — after +trail_after_r, trail the stop under each bar's low
(prior_low) or a % off the high-water mark (pct)
* scale-out — sell scale_out_pct of the position at +scale_out_r, run the rest
* first-red — exit on the first bar that closes below the prior bar's low
* target_r — final take-profit (whatever remains)
Intrabar convention is conservative: within a bar the STOP is assumed to fill
before the target/scale, so the backtest never flatters a trade.
"""
from __future__ import annotations
import os
from dataclasses import dataclass, field
import pandas as pd
TRAIL_NONE = "none"
TRAIL_PRIOR_LOW = "prior_low"
TRAIL_PCT = "pct"
@dataclass
class ExitConfig:
target_r: float = 2.0 # final take-profit in R (0 = no fixed target)
breakeven_at_r: float = 0.0 # move stop to entry once +this R is reached (0 = off)
trail_mode: str = TRAIL_NONE # none | prior_low | pct
trail_pct: float = 0.0 # pct trail: stop = high_water * (1 - trail_pct)
trail_after_r: float = 1.0 # only start trailing once +this R is reached
scale_out_r: float = 0.0 # sell scale_out_pct at +this R (0 = off)
scale_out_pct: float = 0.5 # fraction sold at scale_out_r
first_red_exit: bool = False # exit on first close below the prior bar's low
@classmethod
def from_env(cls, env: dict | None = None) -> "ExitConfig":
v = dict(os.environ)
if env:
v.update(env)
def f(key, default):
try:
return float(v.get(key, default))
except (TypeError, ValueError):
return float(default)
def b(key, default):
return v.get(key, default).strip().lower() in {"1", "true", "yes", "on"}
return cls(
target_r=f("TRADING_REWARD_MULTIPLE", "2.0"),
breakeven_at_r=f("TRADING_EXIT_BREAKEVEN_R", "0.0"),
trail_mode=v.get("TRADING_EXIT_TRAIL_MODE", TRAIL_NONE).strip().lower(),
trail_pct=f("TRADING_EXIT_TRAIL_PCT", "0.0"),
trail_after_r=f("TRADING_EXIT_TRAIL_AFTER_R", "1.0"),
scale_out_r=f("TRADING_EXIT_SCALE_OUT_R", "0.0"),
scale_out_pct=f("TRADING_EXIT_SCALE_OUT_PCT", "0.5"),
first_red_exit=b("TRADING_EXIT_FIRST_RED", "0"),
)
def describe(self) -> str:
parts = [f"target {self.target_r:g}R"]
if self.breakeven_at_r:
parts.append(f"BE@{self.breakeven_at_r:g}R")
if self.trail_mode != TRAIL_NONE:
how = f"{self.trail_pct:.1%}" if self.trail_mode == TRAIL_PCT else "prior-low"
parts.append(f"trail {how} after {self.trail_after_r:g}R")
if self.scale_out_r:
parts.append(f"scale {self.scale_out_pct:.0%}@{self.scale_out_r:g}R")
if self.first_red_exit:
parts.append("first-red")
return ", ".join(parts)
@dataclass
class ExitFill:
frac: float # fraction of the original position closed by this fill
price: float
reason: str
@dataclass
class ExitResult:
r_multiple: float # realized R, weighted across partial fills
reason: str # reason of the FINAL (position-closing) fill
exit_index: int # index in bars_after where the position fully closed
fills: list[ExitFill] = field(default_factory=list)
def _trail_stop(stop: float, entry: float, high_water: float,
bar_low: float, reached_r: float, cfg: ExitConfig) -> float:
"""The stop after processing one bar — only ever ratchets UP."""
new = stop
if cfg.breakeven_at_r and reached_r >= cfg.breakeven_at_r:
new = max(new, entry)
if reached_r >= cfg.trail_after_r:
if cfg.trail_mode == TRAIL_PRIOR_LOW:
new = max(new, bar_low)
elif cfg.trail_mode == TRAIL_PCT and cfg.trail_pct > 0:
new = max(new, high_water * (1.0 - cfg.trail_pct))
return new
def simulate_exit(
entry_price: float, init_stop: float, bars_after: pd.DataFrame, cfg: ExitConfig
) -> ExitResult:
"""Walk the bars AFTER entry and return the realized R for the whole trade.
``bars_after`` are the OHLC bars strictly after the entry bar, in order.
The same per-bar rules are used live (see ``manage_live``).
"""
risk = entry_price - init_stop
if risk <= 0 or bars_after is None or len(bars_after) == 0:
return ExitResult(0.0, "invalid", 0, [])
target = entry_price + cfg.target_r * risk if cfg.target_r else None
stop = init_stop
high_water = entry_price
remaining = 1.0
realized_r = 0.0
scaled = False
fills: list[ExitFill] = []
prev_low: float | None = None
n = len(bars_after)
for idx in range(n):
bar = bars_after.iloc[idx]
high = float(bar["high"]); low = float(bar["low"]); close = float(bar["close"])
high_water = max(high_water, high)
reached_r = (high_water - entry_price) / risk
# (1) stop first (conservative intrabar ordering)
if low <= stop:
r = (stop - entry_price) / risk
realized_r += remaining * r
reason = "breakeven" if abs(stop - entry_price) < 1e-9 else (
"trail_stop" if stop > init_stop else "stop_loss")
fills.append(ExitFill(remaining, stop, reason))
return ExitResult(realized_r, reason, idx, fills)
# (2) scale-out into strength
if (cfg.scale_out_r and not scaled
and high >= entry_price + cfg.scale_out_r * risk):
frac = min(cfg.scale_out_pct, remaining)
realized_r += frac * cfg.scale_out_r
remaining -= frac
scaled = True
fills.append(ExitFill(frac, entry_price + cfg.scale_out_r * risk, "scale_out"))
if remaining <= 1e-9:
return ExitResult(realized_r, "scale_out", idx, fills)
# (3) final target on the remainder
if target is not None and high >= target:
realized_r += remaining * cfg.target_r
fills.append(ExitFill(remaining, target, "target"))
return ExitResult(realized_r, "target", idx, fills)
# (4) first red candle that breaks the prior bar's low
if cfg.first_red_exit and prev_low is not None and close < prev_low:
r = (close - entry_price) / risk
realized_r += remaining * r
fills.append(ExitFill(remaining, close, "first_red"))
return ExitResult(realized_r, "first_red", idx, fills)
# (5) ratchet the stop for the NEXT bar
stop = _trail_stop(stop, entry_price, high_water, low, reached_r, cfg)
prev_low = low
# session end on the remainder
last_close = float(bars_after.iloc[-1]["close"])
realized_r += remaining * ((last_close - entry_price) / risk)
fills.append(ExitFill(remaining, last_close, "session_end"))
return ExitResult(realized_r, "session_end", n - 1, fills)
@dataclass
class ExitDecision:
desired_stop: float # where the protective stop SHOULD be now
scale_out_frac: float # fraction to sell at market now (0 if none)
exit_now: bool # full market exit now (first-red)
reason: str
def manage_live(
entry_price: float, init_stop: float, bars_since_entry: pd.DataFrame,
cfg: ExitConfig, already_scaled: bool = False,
) -> ExitDecision:
"""Compute the management action for an OPEN live position.
Uses the same ratchet/scale/first-red rules as ``simulate_exit`` so the live
behaviour matches the backtest. ``bars_since_entry`` are the closed bars from
the entry bar onward. Returns the stop the broker order should be moved to
(only ever up), whether to scale out now, and whether to exit outright.
"""
risk = entry_price - init_stop
if risk <= 0 or bars_since_entry is None or len(bars_since_entry) == 0:
return ExitDecision(init_stop, 0.0, False, "hold")
stop = init_stop
high_water = entry_price
for idx in range(len(bars_since_entry)):
bar = bars_since_entry.iloc[idx]
high_water = max(high_water, float(bar["high"]))
reached_r = (high_water - entry_price) / risk
stop = _trail_stop(stop, entry_price, high_water, float(bar["low"]),
reached_r, cfg)
last = bars_since_entry.iloc[-1]
last_close = float(last["close"])
reached_r = (high_water - entry_price) / risk
# first-red: latest bar closed below the prior bar's low
if cfg.first_red_exit and len(bars_since_entry) >= 2:
prior_low = float(bars_since_entry.iloc[-2]["low"])
if last_close < prior_low and reached_r >= cfg.trail_after_r:
return ExitDecision(stop, 0.0, True, "first_red")
# scale-out: high reached the scale level and we haven't yet
scale_frac = 0.0
if (cfg.scale_out_r and not already_scaled
and high_water >= entry_price + cfg.scale_out_r * risk):
scale_frac = cfg.scale_out_pct
return ExitDecision(stop, scale_frac, False, "trail")