98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
from datetime import date
|
|
|
|
from scripts.run_daily_reentry_matrix import PrecomputedDailyEngine, ReentryPolicy
|
|
|
|
|
|
RANKING_KEY = "strategy_rank"
|
|
ORD = date(2025, 1, 6).toordinal()
|
|
|
|
|
|
def _candidate(day_ord: int, *, stop: float = 91.0, rank: float = 81.0) -> dict:
|
|
return {
|
|
"qualified": True,
|
|
"direction": "long",
|
|
"symbol": "AAA",
|
|
"date": date.fromordinal(day_ord).isoformat(),
|
|
"entry": 100.0,
|
|
"stop": stop,
|
|
"target": 120.0,
|
|
RANKING_KEY: rank,
|
|
}
|
|
|
|
|
|
def _state(sessions: int = 0) -> dict:
|
|
return {
|
|
"sessions_since_stop": sessions,
|
|
"previous_stop": 90.0,
|
|
"previous_rank": 80.0,
|
|
"gate_went_unqualified": False,
|
|
}
|
|
|
|
|
|
def _call(policy: ReentryPolicy, day_ord: int, state: dict, sessions: int):
|
|
state["sessions_since_stop"] = sessions
|
|
return policy("AAA", day_ord, state, object())
|
|
|
|
|
|
def test_next_session_blocks_only_stop_day():
|
|
engine = PrecomputedDailyEngine([_candidate(ORD), _candidate(ORD + 1)])
|
|
policy = ReentryPolicy("next_session", engine, RANKING_KEY)
|
|
state = _state()
|
|
|
|
assert _call(policy, ORD, state, 0) is None
|
|
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)
|
|
state = _state()
|
|
|
|
assert _call(policy, ORD + 4, state, 4) is None
|
|
assert _call(policy, ORD + 5, state, 5) is not None
|
|
|
|
|
|
def test_gate_reset_requires_failure_before_requalification():
|
|
engine = PrecomputedDailyEngine([_candidate(ORD), _candidate(ORD + 2)])
|
|
policy = ReentryPolicy("gate_reset", engine, RANKING_KEY)
|
|
state = _state()
|
|
|
|
assert _call(policy, ORD, state, 0) is None
|
|
assert _call(policy, ORD + 1, state, 1) is None
|
|
emitted = _call(policy, ORD + 2, state, 2)
|
|
assert emitted is not None
|
|
assert emitted["_reentry_reason"] == "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),
|
|
_candidate(ORD + 2, stop=92.0, rank=79.0),
|
|
_candidate(ORD + 3, stop=92.0, rank=81.0),
|
|
])
|
|
policy = ReentryPolicy("gate_reset_improved", engine, RANKING_KEY)
|
|
state = _state()
|
|
|
|
assert _call(policy, ORD, state, 0) is None
|
|
assert _call(policy, ORD + 1, state, 1) is None
|
|
assert _call(policy, ORD + 2, state, 2) is None
|
|
emitted = _call(policy, ORD + 3, state, 3)
|
|
assert emitted is not None
|
|
assert emitted["_reentry_reason"] == "gate_reset_with_improved_stop_and_rank"
|
|
|
|
|
|
def test_two_session_confirmation_excludes_stop_day_close():
|
|
engine = PrecomputedDailyEngine([
|
|
_candidate(ORD),
|
|
_candidate(ORD + 1),
|
|
_candidate(ORD + 2),
|
|
])
|
|
policy = ReentryPolicy("two_session_confirmation", engine, RANKING_KEY)
|
|
state = _state()
|
|
|
|
assert _call(policy, ORD, state, 0) is None
|
|
assert _call(policy, ORD + 1, state, 1) is None
|
|
emitted = _call(policy, ORD + 2, state, 2)
|
|
assert emitted is not None
|
|
assert emitted["_reentry_reason"] == "two_qualified_post_stop_closes"
|