You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
"""Unit tests for live ORB opening-burst entry parity."""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from apps.orb_trader.engine import ORBTradingEngine
|
|
|
|
_DATE = "2026-01-05"
|
|
_ET = ZoneInfo("America/New_York")
|
|
|
|
|
|
def _make_engine() -> ORBTradingEngine:
|
|
params = SimpleNamespace(
|
|
sim_bar_minutes=5,
|
|
orb_minutes=5,
|
|
order_timeout_minutes=25,
|
|
live_breakout_use_bar_high=False,
|
|
market_thrust_liquid_continuation_entry_mode="opening_burst",
|
|
opening_burst_liquid_max_entry_minutes_after_open=15,
|
|
daily_budget_reset=True,
|
|
drawdown_governor_threshold=None,
|
|
drawdown_governor_min_scale=0.30,
|
|
streak_sizing_win_bonus=None,
|
|
streak_sizing_loss_penalty=None,
|
|
streak_sizing_max=2.5,
|
|
streak_sizing_min=0.5,
|
|
max_simultaneous_entries=3,
|
|
atr_stop_multiplier=0.75,
|
|
risk_per_trade_pct=0.05,
|
|
max_position_pct=0.70,
|
|
)
|
|
session = SimpleNamespace(
|
|
session_id="test-session",
|
|
session_name="test",
|
|
initial_equity=10_000.0,
|
|
)
|
|
state = MagicMock()
|
|
state.get_open_positions.return_value = []
|
|
state.get_equity.return_value = 10_000.0
|
|
state.get_peak_equity.return_value = 10_000.0
|
|
state.get_daily_state.return_value = SimpleNamespace(kill_switch=False)
|
|
|
|
broker = MagicMock()
|
|
broker.get_account.return_value = SimpleNamespace(buying_power=100_000.0)
|
|
broker.submit_market_buy.return_value = SimpleNamespace(id="buy-1")
|
|
broker.get_order.return_value = SimpleNamespace(
|
|
status="filled",
|
|
filled_avg_price=100.50,
|
|
)
|
|
|
|
eng = object.__new__(ORBTradingEngine)
|
|
eng._session = session
|
|
eng._params = params
|
|
eng._state = state
|
|
eng._broker = broker
|
|
eng._log_callback = None
|
|
eng._date_str = _DATE
|
|
eng._day_size_scale = 1.0
|
|
eng._market_orb_quality_max_trades = None
|
|
eng._market_orb_quality_reason = None
|
|
eng._pending_cands = [
|
|
{
|
|
"ticker": "BURST",
|
|
"direction": "bullish",
|
|
"orb_bar": {
|
|
"timestamp": f"{_DATE}T09:30:00-05:00",
|
|
"open": 100.0,
|
|
"high": 105.0,
|
|
"low": 99.0,
|
|
"close": 104.0,
|
|
"volume": 1_000,
|
|
},
|
|
"atr": 2.0,
|
|
"rvol": 6.0,
|
|
"gap_pct": 0.08,
|
|
"score": 0.95,
|
|
"market_thrust_liquid_continuation": True,
|
|
}
|
|
]
|
|
eng._candidates = list(eng._pending_cands)
|
|
eng._now_et = lambda: dt.datetime(2026, 1, 5, 9, 40, tzinfo=_ET) # type: ignore[method-assign]
|
|
return eng
|
|
|
|
|
|
@patch("apps.orb_trader.engine.time")
|
|
@patch("apps.orb_trader.engine.get_snapshots")
|
|
def test_market_thrust_opening_burst_enters_without_snapshot_breakout(
|
|
mock_snapshots,
|
|
mock_time,
|
|
):
|
|
mock_snapshots.return_value = {"BURST": SimpleNamespace(price=100.0)}
|
|
eng = _make_engine()
|
|
|
|
result = eng.run_breakout_check(_DATE)
|
|
|
|
assert result["filled"] == 1
|
|
eng._broker.submit_market_buy.assert_called_once()
|
|
pos = eng._state.save_position.call_args.args[0]
|
|
assert pos.ticker == "BURST"
|
|
assert pos.trigger_type == "market_thrust_opening_burst"
|
|
assert pos.entry_price == 100.50
|