import math import sys from pathlib import Path from types import SimpleNamespace import pytest ROOT = Path("/mnt/dolphinng5_predict") sys.path.insert(0, str(ROOT / "nautilus_dolphin")) sys.path.insert(1, str(ROOT)) if "nautilus_dolphin" in sys.modules: pkg = sys.modules["nautilus_dolphin"] pkg_file = str(getattr(pkg, "__file__", "") or "") if not pkg_file.endswith("nautilus_dolphin/nautilus_dolphin/__init__.py"): del sys.modules["nautilus_dolphin"] from nautilus_dolphin.nautilus.adaptive_circuit_breaker import AdaptiveCircuitBreaker, ACBConfig from nautilus_dolphin.nautilus.alpha_bet_sizer import AlphaBetSizer from nautilus_dolphin.nautilus.alpha_exit_manager import AlphaExitManager from nautilus_dolphin.nautilus.alpha_signal_generator import AlphaSignalGenerator from nautilus_dolphin.nautilus.esf_alpha_orchestrator import NDAlphaEngine from nautilus_dolphin.nautilus.dolphin_actor import _trade_direction_from_config def test_signal_generator_long_gate_and_dc_are_side_aware(): gen = AlphaSignalGenerator(use_direction_confirm=True) rising_prices = [100.0, 100.1, 100.2, 100.3, 100.4, 100.5, 100.6, 101.0] falling_prices = [101.0, 100.8, 100.6, 100.4, 100.2, 100.0, 99.8, 99.5] long_sig = gen.generate( vel_div=0.025, vel_div_history=[0.012] * 10, asset_price_history=rising_prices, trade_direction=1, ) assert long_sig.is_valid assert long_sig.direction == 1 assert long_sig.dc_status == "CONFIRM" contradicted = gen.generate( vel_div=0.025, vel_div_history=[0.012] * 10, asset_price_history=falling_prices, trade_direction=1, ) assert not contradicted.is_valid assert contradicted.dc_status == "SKIP_CONTRADICT" def test_bet_sizer_trend_multiplier_is_direction_aware_for_long(): sizer = AlphaBetSizer( base_fraction=0.20, min_leverage=0.5, max_leverage=8.0, use_alpha_layers=True, use_dynamic_leverage=True, ) favorable_long = sizer.calculate_size(25000, 0.025, vel_div_trend=0.02, trade_direction=1) adverse_long = sizer.calculate_size(25000, 0.025, vel_div_trend=-0.02, trade_direction=1) favorable_short = sizer.calculate_size(25000, -0.035, vel_div_trend=-0.02, trade_direction=-1) adverse_short = sizer.calculate_size(25000, -0.035, vel_div_trend=0.02, trade_direction=-1) assert favorable_long["fraction"] > adverse_long["fraction"] assert favorable_short["fraction"] > adverse_short["fraction"] def test_ndalphaengine_enters_long_when_begin_day_direction_is_long(): engine = NDAlphaEngine( initial_capital=1000.0, use_asset_selection=False, use_direction_confirm=False, use_sp_fees=False, use_sp_slippage=False, use_ob_edge=False, use_alpha_layers=False, use_dynamic_leverage=False, lookback=1, ) engine.begin_day("2026-05-08", posture="APEX", direction=1) res = engine.step_bar(0, vel_div=0.025, prices={"BTCUSDT": 100.0}, vol_regime_ok=True) assert res["entry"] is not None assert res["entry"]["direction"] == 1 assert engine.position is not None assert engine.position.direction == 1 def test_ndalphaengine_short_default_preserved(): engine = NDAlphaEngine( initial_capital=1000.0, use_asset_selection=False, use_direction_confirm=False, use_sp_fees=False, use_sp_slippage=False, use_ob_edge=False, use_alpha_layers=False, use_dynamic_leverage=False, lookback=1, ) engine.begin_day("2026-05-08", posture="APEX") res = engine.step_bar(0, vel_div=-0.035, prices={"BTCUSDT": 100.0}, vol_regime_ok=True) assert res["entry"] is not None assert res["entry"]["direction"] == -1 def test_acb_short_default_and_long_cache_are_side_separated(): acb = AdaptiveCircuitBreaker() acb._w750_threshold = 0.001 bullish = { "funding_btc": 0.0002, "dvol_btc": 30.0, "fng": 80.0, "taker": 1.25, "available": True, } short = acb._calculate_signals(bullish) long = acb._calculate_signals(bullish, direction=1) assert short["signals"] == pytest.approx(0.0) assert long["signals"] == pytest.approx(4.0) snap = dict(bullish, _acb_ready=True, _staleness_s={}) short_hz = acb.get_dynamic_boost_from_hz("2026-05-08", snap, w750_velocity=0.002, direction=-1) long_hz = acb.get_dynamic_boost_from_hz("2026-05-08", snap, w750_velocity=0.002, direction=1) assert short_hz["side"] == "SHORT" assert long_hz["side"] == "LONG" assert short_hz["boost"] == pytest.approx(1.0) assert long_hz["boost"] == pytest.approx(1.0 + 0.5 * math.log1p(4.0)) assert acb.get_dynamic_boost_for_date("2026-05-08")["side"] == "SHORT" assert acb.get_dynamic_boost_for_date("2026-05-08", direction=1)["side"] == "LONG" def test_acb_short_threshold_regression_values_still_match_v6(): acb = AdaptiveCircuitBreaker() factors = { "funding_btc": -0.0002, "dvol_btc": 85.0, "fng": 20.0, "taker": 0.75, "available": True, } result = acb._calculate_signals(factors) assert result["signals"] == pytest.approx(4.0) assert result["severity"] == 7 def test_acb_ob_beta_modulation_is_side_aware(): acb = AdaptiveCircuitBreaker() acb._w750_threshold = 0.001 calm_ob = SimpleNamespace( get_macro=lambda: SimpleNamespace(regime_signal=-1, depth_velocity=0.1, cascade_count=0) ) stress_ob = SimpleNamespace( get_macro=lambda: SimpleNamespace(regime_signal=1, depth_velocity=-0.3, cascade_count=2) ) long_calm = acb.get_dynamic_boost_from_hz( "2026-05-08", {"_acb_ready": True, "_staleness_s": {}}, w750_velocity=0.002, ob_engine=calm_ob, direction=1 ) short_calm = acb.get_dynamic_boost_from_hz( "2026-05-09", {"_acb_ready": True, "_staleness_s": {}}, w750_velocity=0.002, ob_engine=calm_ob, direction=-1 ) short_stress = acb.get_dynamic_boost_from_hz( "2026-05-10", {"_acb_ready": True, "_staleness_s": {}}, w750_velocity=0.002, ob_engine=stress_ob, direction=-1 ) assert long_calm["beta"] == pytest.approx(1.0) assert short_calm["beta"] == pytest.approx(0.68) assert short_stress["beta"] == pytest.approx(1.0) def test_exit_manager_optional_vd_exit_is_long_aware(): manager = AlphaExitManager(vd_enabled=True, vd_consec_bars=2) manager.setup_position("long-1", entry_price=100.0, direction=1, entry_bar=0) first = manager.evaluate("long-1", current_price=100.1, current_bar=1, vel_div=-0.02) second = manager.evaluate("long-1", current_price=100.1, current_bar=2, vel_div=-0.02) assert first["action"] == "HOLD" assert second["action"] == "EXIT" assert second["reason"] == "VD_INVALIDATION" def test_prodgreen_direction_parser_is_explicit_and_case_insensitive(): assert _trade_direction_from_config("LONG_ONLY") == 1 assert _trade_direction_from_config("short_only") == -1 with pytest.raises(ValueError): _trade_direction_from_config("bidirectional")