dennisthiessen/signal-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add five-session post-stop reentry lockdown

Browse Source
This commit is contained in:
dennisthiessen
2026-07-17 13:21:06 +02:00
parent f714782fa4
commit 1e9f2dc4fb
14 changed files with 36919 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_backtest_service.py
+169
View File
@@ -575,6 +575,170 @@ class TestSimulatePortfolio:
assert sim["trades"] == 1
assert sim["worst_trade_r"] == pytest.approx(-2.0) # (90 − 100) / 5
def test_initial_stop_cooldown_blocks_immediate_reentry(self):
closes = [100.0, 94.0, 96.0]
prices = {"AAA": _sim_prices(self.ORD, closes)}
candidates = [
_sim_cand("AAA", self.ORD, entry=100.0, stop=95.0, target=120.0),
_sim_cand("AAA", self.ORD + 1, entry=94.0, stop=89.0, target=110.0),
]
baseline = bt._simulate_portfolio(candidates, prices, None, "hold", 30)
cooldown = bt._simulate_portfolio(
candidates,
prices,
None,
"hold",
30,
reentry_cooldown_days=5,
)
assert baseline is not None and baseline["trades"] == 2
assert cooldown is not None and cooldown["trades"] == 1
assert cooldown["skipped_cooldown"] == 1
assert cooldown["reentry_cooldown_days"] == 5
def test_production_monitor_applies_live_reentry_lockdown(self, monkeypatch):
def fake_simulator(*_args, **kwargs):
return {
"trades": 0,
"applied_reentry_lockdown": kwargs.get("reentry_cooldown_days", 0),
}
monkeypatch.setattr(bt, "_simulate_portfolio", fake_simulator)
market_ord = date(2026, 7, 1).toordinal()
prices = {"AAA": ([market_ord], [], [], [], [], [])}
monitor = bt._portfolio_monitor([], prices, None, 30)
production_rows = [
row for row in monitor["runs"] if row["is_production"]
]
comparison_rows = [
row for row in monitor["runs"] if not row["is_production"]
]
assert production_rows
assert all(
row["reentry_lockdown_sessions"] == bt.REENTRY_LOCKDOWN_SESSIONS
and row["applied_reentry_lockdown"] == bt.REENTRY_LOCKDOWN_SESSIONS
for row in production_rows
)
assert comparison_rows
assert all(
row["reentry_lockdown_sessions"] == 0
and row["applied_reentry_lockdown"] == 0
for row in comparison_rows
)
def test_initial_stop_can_refresh_lower_and_survive_same_bar(self):
closes = [100.0, 94.0, 96.0]
prices = {"AAA": _sim_prices(self.ORD, closes)}
candidate = _sim_cand(
"AAA", self.ORD, entry=100.0, stop=95.0, target=120.0
)
sim = bt._simulate_portfolio(
[candidate],
prices,
None,
"hold",
2,
initial_stop_refresh_fn=lambda *_: 90.0,
include_trades=True,
)
assert sim is not None
assert sim["stop_refresh_attempts"] == 1
assert sim["stop_refreshes"] == 1
assert sim["stop_refresh_same_bar_hits"] == 0
assert sim["exit_reasons"] == {"time": 1}
assert sim["trade_details"][0]["stop_refreshes"] == 1
def test_refreshed_stop_is_checked_against_same_bar(self):
ords = list(range(self.ORD, self.ORD + 2))
prices = {
"AAA": (
ords,
[100.0, 94.0],
[101.0, 96.0],
[99.0, 89.0],
[100.0, 94.0],
[1, 1],
)
}
candidate = _sim_cand(
"AAA", self.ORD, entry=100.0, stop=95.0, target=120.0
)
sim = bt._simulate_portfolio(
[candidate],
prices,
None,
"hold",
30,
initial_stop_refresh_fn=lambda *_: 90.0,
)
assert sim is not None
assert sim["stop_refresh_same_bar_hits"] == 1
assert sim["worst_trade_r"] == pytest.approx(-2.0)
def test_post_stop_state_suppresses_same_episode_candidate(self):
closes = [100.0, 94.0, 96.0]
prices = {"AAA": _sim_prices(self.ORD, closes)}
candidates = [
_sim_cand("AAA", self.ORD, entry=100.0, stop=95.0, target=120.0),
_sim_cand("AAA", self.ORD + 1, entry=94.0, stop=89.0, target=110.0),
]
sim = bt._simulate_portfolio(
candidates,
prices,
None,
"hold",
30,
post_stop_reentry_fn=lambda *_: None,
)
assert sim is not None
assert sim["trades"] == 1
assert sim["post_stop_events"] == 1
assert sim["post_stop_reentries"] == 0
assert sim["post_stop_states_open_at_end"] == 1
def test_post_stop_callback_can_reenter_same_day(self):
closes = [100.0, 94.0, 96.0]
prices = {"AAA": _sim_prices(self.ORD, closes)}
initial = _sim_cand(
"AAA", self.ORD, entry=100.0, stop=95.0, target=120.0
)
def immediate_reentry(sym, current_ord, _state, bar):
return _sim_cand(
sym,
current_ord,
entry=bar.close,
stop=bar.close - 5.0,
target=bar.close + 15.0,
)
sim = bt._simulate_portfolio(
[initial],
prices,
None,
"hold",
30,
post_stop_reentry_fn=immediate_reentry,
include_trades=True,
)
assert sim is not None
assert sim["trades"] == 2
assert sim["post_stop_reentries"] == 1
assert sim["reentry_events"][0]["wait_sessions"] == 0
assert sim["trade_details"][1]["is_reentry"] is True
assert sim["trade_details"][1]["reentry_wait_sessions"] == 0
def test_sma50_policy_exits_on_close_break(self):
closes = [100.0] * 56 + [90.0, 91.0]
prices = {"AAA": _sim_prices(self.ORD, closes)}
@@ -722,6 +886,7 @@ def test_build_recommendation_prefers_production_monitor_headline():
})
assert rec["headline"] is not None
assert "3x ATR trailing exit" in rec["headline"]
assert "5-session re-entry lockdown" in rec["headline"]
assert any(item["topic"] == "production" for item in rec["items"])
@@ -870,6 +1035,10 @@ async def test_run_backtest_smoke(session):
assert report["params"]["cost_per_side_pct"] == pytest.approx(bt.COST_PER_SIDE * 100)
assert report["params"]["target_model"] == bt.PRODUCTION_GTL_TARGET_MODEL
assert report["params"]["is_production_target_model"] is True
assert (
report["params"]["production_reentry_lockdown_sessions"]
== bt.REENTRY_LOCKDOWN_SESSIONS
)
assert "net_avg_r" in report["overall_all"]
# ablation baseline reproduces the qualified set exactly, and every row
tests/unit/test_rr_scanner_preservation.py
+93
View File
@@ -607,6 +607,99 @@ async def test_get_trade_setups_can_exclude_tickers_with_open_paper_trades(
assert [row["symbol"] for row in ticker_rows] == ["OPENQ"]
@pytest.mark.asyncio
async def test_get_trade_setups_applies_five_session_initial_stop_lockdown(
db_session: AsyncSession,
):
now = datetime.now(timezone.utc)
today = now.date()
if await db_session.get(User, 1) is None:
db_session.add(
User(id=1, username="u", password_hash="x", role="user", has_access=True)
)
await db_session.flush()
blocked = Ticker(symbol="STOP4")
released = Ticker(symbol="STOP5")
trailing = Ticker(symbol="TRAILQ")
db_session.add_all([blocked, released, trailing])
await db_session.flush()
# Six synthetic stored market sessions D0..D5. A stop on D0 has five
# later sessions and is released; a stop on D1 has only four and is not.
market_sessions = [today - timedelta(days=offset) for offset in range(5, -1, -1)]
for market_date in market_sessions:
db_session.add(
OHLCVRecord(
ticker_id=blocked.id,
date=market_date,
open=100.0,
high=101.0,
low=99.0,
close=100.0,
volume=1_000,
)
)
for ticker in (blocked, released, trailing):
db_session.add(
TradeSetup(
ticker_id=ticker.id,
direction="long",
entry_price=100.0,
stop_loss=95.0,
target=115.0,
rr_ratio=3.0,
composite_score=80.0,
detected_at=now,
)
)
def closed_trade(ticker: Ticker, closed_on: date, reason: str) -> PaperTrade:
return PaperTrade(
user_id=1,
ticker_id=ticker.id,
direction="long",
entry_price=100.0,
shares=10.0,
stop_loss=95.0,
target=115.0,
status="closed",
opened_at=datetime.combine(
closed_on - timedelta(days=1), datetime.min.time(), tzinfo=timezone.utc
),
close_price=95.0,
closed_at=datetime.combine(
closed_on, datetime.min.time(), tzinfo=timezone.utc
),
close_reason=reason,
)
db_session.add_all(
[
closed_trade(blocked, market_sessions[1], "stop"),
closed_trade(released, market_sessions[0], "stop"),
closed_trade(trailing, market_sessions[-1], "trailing"),
]
)
await db_session.flush()
default_symbols = {
row["symbol"] for row in await get_trade_setups(db_session)
}
assert {"STOP4", "STOP5", "TRAILQ"}.issubset(default_symbols)
available_symbols = {
row["symbol"]
for row in await get_trade_setups(
db_session,
exclude_reentry_lockdown_tickers=True,
)
}
assert "STOP4" not in available_symbols
assert {"STOP5", "TRAILQ"}.issubset(available_symbols)
async def _seed_stale_setup_with_current_scores(db_session: AsyncSession) -> TradeSetup:
"""Stored setup frozen at scan time (conf 82, neutral) vs. current context
(bullish sentiment, composite 96) that yields live confidence 97.