feat: test shorter post-stop reentry guards

This commit is contained in:
2026-07-17 17:37:03 +02:00
parent cf294a7d2b
commit 5385f46064
2 changed files with 51 additions and 11 deletions
+17 -5
View File
@@ -32,8 +32,11 @@ if str(ROOT) not in sys.path:
POLICY_NAMES = (
"immediate",
"next_session",
"cooldown_2",
"cooldown_3",
"cooldown_5",
"gate_reset",
"strict_gate_reset",
"gate_reset_improved",
"two_session_confirmation",
)
@@ -226,6 +229,10 @@ class ReentryPolicy:
sessions = int(state["sessions_since_stop"])
candidate = self.engine.candidate(symbol, asof_ord)
if candidate is None:
# The strict reset deliberately ignores the stop-day gate state:
# it requires a later completed session to go unqualified before
# any requalification can trigger a new entry.
if self.name != "strict_gate_reset" or sessions >= 1:
state["gate_went_unqualified"] = True
state["qualified_streak"] = 0
return None
@@ -244,12 +251,16 @@ class ReentryPolicy:
elif self.name == "next_session":
if sessions >= 1:
reason = "stop_day_block_complete"
elif self.name == "cooldown_5":
if sessions >= 5:
reason = "5_session_cooldown_complete"
elif self.name.startswith("cooldown_"):
cooldown_sessions = int(self.name.removeprefix("cooldown_"))
if sessions >= cooldown_sessions:
reason = f"{cooldown_sessions}_session_cooldown_complete"
elif self.name == "gate_reset":
if state["gate_went_unqualified"]:
reason = "gate_failed_then_requalified"
elif self.name == "strict_gate_reset":
if state["gate_went_unqualified"]:
reason = "post_stop_gate_failed_then_requalified"
elif self.name == "gate_reset_improved":
previous_rank = state.get("previous_rank")
current_rank = candidate.get(self.ranking_key)
@@ -820,8 +831,9 @@ async def _main() -> None:
"same policy, lookback, cost, capacity, and holdout matrix. Each "
"immediate callback must match its direct no-lockdown simulation exactly. "
"Immediate is the daily no-lockdown baseline; next_session blocks only "
"the stop day; cooldown_5 permits re-entry at wait_sessions=5; gate_reset "
"requires an unqualified close before requalification; "
"the stop day; cooldown_N permits re-entry at wait_sessions=N; gate_reset "
"counts the stop-day gate state, while strict_gate_reset requires an "
"unqualified close on a later completed session before requalification; "
"gate_reset_improved additionally requires a higher stop and a non-weaker "
"production rank; two_session_confirmation requires two consecutive "
"qualified post-stop closes and excludes the stop day's close. Transaction "
+33 -5
View File
@@ -49,13 +49,17 @@ def test_next_session_blocks_only_stop_day():
assert _call(policy, ORD + 1, state, 1) is not None
def test_five_session_cooldown_unlocks_at_exact_boundary():
engine = PrecomputedDailyEngine([_candidate(ORD + 4), _candidate(ORD + 5)])
policy = ReentryPolicy("cooldown_5", engine, RANKING_KEY)
@pytest.mark.parametrize("sessions", [2, 3, 5])
def test_cooldown_unlocks_at_exact_boundary(sessions):
engine = PrecomputedDailyEngine([
_candidate(ORD + sessions - 1),
_candidate(ORD + sessions),
])
policy = ReentryPolicy(f"cooldown_{sessions}", engine, RANKING_KEY)
state = _state()
assert _call(policy, ORD + 4, state, 4) is None
assert _call(policy, ORD + 5, state, 5) is not None
assert _call(policy, ORD + sessions - 1, state, sessions - 1) is None
assert _call(policy, ORD + sessions, state, sessions) is not None
def test_gate_reset_requires_failure_before_requalification():
@@ -70,6 +74,30 @@ def test_gate_reset_requires_failure_before_requalification():
assert emitted["_reentry_reason"] == "gate_failed_then_requalified"
def test_strict_gate_reset_ignores_stop_day_failure():
engine = PrecomputedDailyEngine([
_candidate(ORD + 1),
_candidate(ORD + 3),
])
policy = ReentryPolicy("strict_gate_reset", engine, RANKING_KEY)
state = _state()
# An unqualified stop-day close alone does not reset the strict policy.
assert _call(policy, ORD, state, 0) is None
assert state["gate_went_unqualified"] is False
assert _call(policy, ORD + 1, state, 1) is None
# A later unqualified close establishes the reset; only then may the next
# qualified setup re-enter.
assert _call(policy, ORD + 2, state, 2) is None
assert state["gate_went_unqualified"] is True
emitted = _call(policy, ORD + 3, state, 3)
assert emitted is not None
assert emitted["_reentry_reason"] == (
"post_stop_gate_failed_then_requalified"
)
def test_improved_gate_reset_requires_better_stop_and_non_weaker_rank():
engine = PrecomputedDailyEngine([
_candidate(ORD + 1, stop=89.0, rank=82.0),