"""Focused tests for the HermesFace Futures integration. Mocked exchange/HTTP only -- never submits a real order. Run with: pytest hermes_overlay/tests -q (run from a checkout where hermes-agent's `trading`/`tools` packages are importable, e.g. inside /opt/hermes, or after `pip install -e .` there). """ import asyncio import os import sys import types import pytest sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) # hermes_overlay/ from trading.symbols import normalize_symbol from trading.risk import calculate_futures_size, RiskError, RISK_PROFILES import trading.dual_datasource_client as ddc import trading.futures_execution as fx # --------------------------------------------------------------------------- # Symbol normalization # --------------------------------------------------------------------------- @pytest.mark.parametrize("raw", ["BTC", "BTCUSDT", "BTC/USDT", "BTC/USDT:USDT", "btc/usdt"]) def test_symbol_normalization(raw): norm = normalize_symbol(raw) assert norm.base == "BTC" assert norm.quote == "USDT" assert norm.ds4 == "BTCUSDT" assert norm.ds2 == "BTC" assert norm.ccxt_perp == "BTC/USDT:USDT" def test_symbol_normalization_rejects_empty(): with pytest.raises(ValueError): normalize_symbol("") # --------------------------------------------------------------------------- # Risk / sizing # --------------------------------------------------------------------------- def test_calculate_futures_size_basic(): result = calculate_futures_size( account_equity=10_000, entry_price=100, stop_loss=98, risk_profile="moderate", requested_leverage=5, ) assert result.risk_amount == 300 # 3% of 10000 assert result.quantity == pytest.approx(150.0) # 300 / 2 assert result.effective_leverage == 5 def test_calculate_futures_size_leverage_capped_by_profile(): result = calculate_futures_size( account_equity=10_000, entry_price=100, stop_loss=98, risk_profile="conservative", requested_leverage=50, ) assert result.effective_leverage == RISK_PROFILES["conservative"]["max_leverage"] assert any("capped" in w for w in result.warnings) def test_calculate_futures_size_high_atr_reduces_leverage(): result = calculate_futures_size( account_equity=10_000, entry_price=100, stop_loss=98, risk_profile="aggressive", requested_leverage=15, atr=5, # 5% of price ) assert result.effective_leverage < 15 assert any("volatility" in w.lower() for w in result.warnings) def test_calculate_futures_size_rejects_missing_stop_distance(): with pytest.raises(RiskError): calculate_futures_size( account_equity=10_000, entry_price=100, stop_loss=100, risk_profile="moderate", requested_leverage=5, ) def test_calculate_futures_size_rejects_below_min_notional(): # account_equity=10, risk 1% -> risk_amount=0.1; stop_distance=1 -> quantity=0.1 # notional = 0.1 * 100 = 10, which is below min_notional=50 with pytest.raises(RiskError): calculate_futures_size( account_equity=10, entry_price=100, stop_loss=99, risk_profile="conservative", requested_leverage=5, min_notional=50, ) def test_calculate_futures_size_max_daily_loss_blocks_trade(): with pytest.raises(RiskError): calculate_futures_size( account_equity=10_000, entry_price=100, stop_loss=98, risk_profile="moderate", requested_leverage=5, max_daily_loss=100, today_realized_loss=150, ) def test_calculate_futures_size_max_concurrent_positions_blocks_trade(): with pytest.raises(RiskError): calculate_futures_size( account_equity=10_000, entry_price=100, stop_loss=98, risk_profile="moderate", requested_leverage=5, open_positions=3, max_concurrent_positions=3, ) # --------------------------------------------------------------------------- # Dual-datasource routing / NO_TRADE guard # --------------------------------------------------------------------------- class _FakeResponse: def __init__(self, payload, status_code=200): self._payload = payload self.status_code = status_code def raise_for_status(self): pass def json(self): return self._payload def _make_fake_client(ds4_payload, ds4_raises=False): class _FakeClient: async def __aenter__(self): return self async def __aexit__(self, *a): return False async def get(self, url, timeout=None, **_kwargs): if "short-hunter/snapshot" in url: if ds4_raises: raise RuntimeError("boom") return _FakeResponse(ds4_payload) return _FakeResponse({}) return _FakeClient() @pytest.fixture(autouse=True) def _disable_live_binance(monkeypatch): async def _unavailable(symbol, needed): return {}, [] monkeypatch.setattr(ddc.binance_public, "get_binance_public_snapshot", _unavailable) def test_no_trade_guard_when_ds4_unreachable(monkeypatch): monkeypatch.setattr(ddc.httpx, "AsyncClient", lambda *a, **kw: _make_fake_client({}, ds4_raises=True)) result = asyncio.run(ddc.get_market_context("BTCUSDT")) assert result["noTradeGuard"] is True assert any("unreachable" in r for r in result["noTradeReasons"]) def test_no_trade_guard_honored_from_ds4(monkeypatch): # Futures fields live under "data" in the real DS4 envelope; only # noTradeGuard/dataState/timestamp are top-level (verified live schema). payload = { "noTradeGuard": True, "dataState": "live", "timestamp": "2026-07-20T11:08:19Z", "data": { "contract": {"x": 1}, "ticker": {"price": 42}, "orderbook": {"asks": [[1, 1]], "bids": [[1, 1]]}, "funding": {"currentFundingRate": 0.0001}, "openInterest": {"openInterest": 100}, }, } monkeypatch.setattr(ddc.httpx, "AsyncClient", lambda *a, **kw: _make_fake_client(payload)) result = asyncio.run(ddc.get_market_context("BTCUSDT")) assert result["noTradeGuard"] is True assert result["sources"]["ticker"] == "datasource4" def test_ds2_cannot_override_ds4_field(monkeypatch): payload = { "noTradeGuard": False, "dataState": "live", "timestamp": "2026-07-20T11:08:19Z", "data": { "contract": {"x": 1}, "ticker": {"price": 42}, "orderbook": {"asks": [[1, 1]], "bids": [[1, 1]]}, "funding": {"currentFundingRate": 0.0001}, "openInterest": {"openInterest": 100}, }, } monkeypatch.setattr(ddc.httpx, "AsyncClient", lambda *a, **kw: _make_fake_client(payload)) result = asyncio.run(ddc.get_market_context("BTCUSDT")) # _normalize_ticker() always adds a canonical lastPrice extracted from # price/last/close/markPrice, in addition to preserving original fields. assert result["merged"]["ticker"] == {"price": 42, "lastPrice": 42.0} assert result["sources"]["ticker"] == "datasource4" # --------------------------------------------------------------------------- # Futures execution safety gates (paper mode; ccxt never touched) # --------------------------------------------------------------------------- @pytest.fixture(autouse=True) def _paper_mode(monkeypatch, tmp_path): monkeypatch.setenv("TRADING_MODE", "paper") monkeypatch.setenv("HERMES_HOME", str(tmp_path)) monkeypatch.delenv("FUTURES_API_KEY", raising=False) monkeypatch.delenv("FUTURES_API_SECRET", raising=False) monkeypatch.delenv(fx.NONPAPER_EXECUTION_ENV, raising=False) # P0-T04: paper functional tests need the kill switch enabled; # dedicated kill-switch tests explicitly disable it. monkeypatch.setenv(fx.EXECUTION_ENABLED_ENV, "true") from trading.domain.schema import apply_migrations assert apply_migrations()[0] yield async def _seed_protected_position( symbol: str = "BTCUSDT", *, side: str = "long", leverage: int = 5, stop_loss: float = 95.0, take_profit: float = 110.0, ): """Test-only setup through the internal service boundary, never a client API.""" from trading.adapters.paper_adapter import PaperExchangeAdapter from trading.domain.execution_service import ExecutionService from trading.domain.identity_repo import ensure_local_paper_account account = ensure_local_paper_account() return await ExecutionService(PaperExchangeAdapter(default_price=100.0))._submit_entry( symbol=symbol, side=side, size=1.0, price=100.0, leverage=leverage, owner_id=account.owner_id, account_id=account.account_id, idempotency_key=f"seed:{symbol}:{side}", stop_loss=stop_loss, take_profit=take_profit, exchange_id="paper", ) def test_trading_mode_forces_paper_without_credentials(monkeypatch): monkeypatch.setenv("TRADING_MODE", "live") assert fx.get_trading_mode() == "paper" def test_trading_mode_forces_paper_without_nonpaper_flag(monkeypatch): """Credentials alone (no HERMES_NONPAPER_EXECUTION_ENABLED) must still force paper.""" monkeypatch.setenv("TRADING_MODE", "live") monkeypatch.setenv("FUTURES_API_KEY", "test-key") monkeypatch.setenv("FUTURES_API_SECRET", "test-secret") monkeypatch.delenv(fx.NONPAPER_EXECUTION_ENV, raising=False) assert fx.get_trading_mode() == "paper" def test_trading_mode_forces_paper_when_nonpaper_flag_falsy(monkeypatch): monkeypatch.setenv("TRADING_MODE", "testnet") monkeypatch.setenv("FUTURES_API_KEY", "test-key") monkeypatch.setenv("FUTURES_API_SECRET", "test-secret") monkeypatch.setenv(fx.NONPAPER_EXECUTION_ENV, "false") assert fx.get_trading_mode() == "paper" def test_trading_mode_honors_live_only_with_all_three_conditions(monkeypatch): """A single env-var change can never bypass containment; all three must be set.""" monkeypatch.setenv("TRADING_MODE", "live") monkeypatch.setenv("FUTURES_API_KEY", "test-key") monkeypatch.setenv("FUTURES_API_SECRET", "test-secret") monkeypatch.setenv(fx.NONPAPER_EXECUTION_ENV, "true") assert fx.get_trading_mode() == "live" def test_trading_mode_single_var_change_cannot_bypass_containment(monkeypatch): """Flipping only TRADING_MODE, with the other two already true/valid, is the one single-variable-change scenario the containment rule must block on its own if either of the other two is absent -- covered by the two tests above. This test additionally confirms testnet behaves identically to live.""" monkeypatch.setenv("TRADING_MODE", "testnet") monkeypatch.delenv("FUTURES_API_KEY", raising=False) monkeypatch.delenv("FUTURES_API_SECRET", raising=False) monkeypatch.setenv(fx.NONPAPER_EXECUTION_ENV, "true") assert fx.get_trading_mode() == "paper" def test_execute_requires_stop_loss(): with pytest.raises(fx.TradingError): asyncio.run(fx.execute_futures_position( symbol="BTCUSDT", side="long", leverage=5, size=1, stop_loss=None, take_profit=110, )) def test_execute_rejects_cross_margin(): with pytest.raises(fx.TradingError): asyncio.run(fx.set_leverage_and_margin("BTCUSDT", 5, margin_type="cross")) def test_execute_rejects_high_slippage(): thin_book = {"asks": [[100, 0.001]], "bids": [[99, 0.001]]} with pytest.raises(fx.TradingError): asyncio.run(fx.execute_futures_position( symbol="BTCUSDT", side="long", leverage=5, size=10, stop_loss=95, take_profit=110, orderbook=thin_book, )) def test_raw_parameter_entry_compatibility_is_hard_rejected(): book = {"asks": [[100, 1000]], "bids": [[99, 1000]]} with pytest.raises(fx.TradingError, match="direct raw futures execution is disabled"): asyncio.run(fx.execute_futures_position( symbol="BTCUSDT", side="long", leverage=5, size=1, stop_loss=95, take_profit=110, orderbook=book, idempotency_key="k1", )) assert asyncio.run(fx.get_futures_positions())["positions"] == [] def test_raw_parameter_short_entry_is_also_rejected(): book = {"asks": [[100, 1000]], "bids": [[99, 1000]]} with pytest.raises(fx.TradingError, match="approved-plan execution API"): asyncio.run(fx.execute_futures_position( symbol="ETHUSDT", side="short", leverage=3, size=1, stop_loss=105, take_profit=90, orderbook=book, )) # --------------------------------------------------------------------------- # P0-T03: LLM agent must not have raw exchange mutation tools # --------------------------------------------------------------------------- def _load_futures_trading_tool_module(): """Load futures_trading_tool without requiring full Hermes tools.registry at import.""" import importlib.util import pathlib path = pathlib.Path(__file__).resolve().parents[1] / "tools" / "futures_trading_tool.py" if "tools" not in sys.modules: tools_pkg = types.ModuleType("tools") sys.modules["tools"] = tools_pkg if "tools.registry" not in sys.modules: reg_mod = types.ModuleType("tools.registry") class _FakeRegistry: def register(self, **kwargs): return None reg_mod.registry = _FakeRegistry() sys.modules["tools.registry"] = reg_mod sys.modules["tools"].registry = reg_mod # type: ignore spec = importlib.util.spec_from_file_location("futures_trading_tool_p0t03", path) mod = importlib.util.module_from_spec(spec) assert spec and spec.loader spec.loader.exec_module(mod) return mod def test_p0_t03_agent_mutation_handlers_reject(): """Handlers for execute/close/set_leverage must hard-reject (P0-T03).""" ftt = _load_futures_trading_tool_module() out = asyncio.run(ftt._h_execute_futures_position({ "symbol": "BTCUSDT", "side": "long", "leverage": 5, "size": 1.0, "stop_loss": 90.0, "take_profit": 120.0, })) assert "NO_TRADE" in out assert "P0-T03" in out or "not available to the LLM agent" in out out2 = asyncio.run(ftt._h_close_futures_position({"symbol": "BTCUSDT"})) assert "NO_TRADE" in out2 out3 = asyncio.run(ftt._h_set_leverage_and_margin({ "symbol": "BTCUSDT", "leverage": 10, })) assert "NO_TRADE" in out3 def test_p0_t03_run_futures_cycle_forces_execute_false(monkeypatch): """Agent run_futures_cycle must ignore execute=True and force analysis-only.""" import importlib.util from pathlib import Path tools_path = Path(__file__).resolve().parents[1] / "plugins" / "futures_trading" / "tools.py" spec = importlib.util.spec_from_file_location("futures_plugin_tools_p0t03", tools_path) tools_mod = importlib.util.module_from_spec(spec) assert spec.loader is not None spec.loader.exec_module(tools_mod) captured = {} async def _fake_cycle(**kwargs): captured.update(kwargs) return { "symbol": kwargs.get("symbol"), "decision": "NO_TRADE", "executed": False, "rejection_reasons": ["analysis only"], } monkeypatch.setattr(tools_mod, "_run_futures_cycle", _fake_cycle) out = asyncio.run(tools_mod._h_run_futures_cycle({ "symbol": "BTCUSDT", "execute": True, "risk_profile": "moderate", })) assert captured.get("execute") is False assert "agentExecuteForcedFalse" in out or "true" in out.lower() def test_p0_t03_dangerous_tools_not_registered_by_plugin(): """Source-level check: plugin register() omits the three mutation tools.""" import pathlib init_path = pathlib.Path(__file__).resolve().parents[1] / "plugins" / "futures_trading" / "__init__.py" text = init_path.read_text(encoding="utf-8") assert 'name="execute_futures_position"' not in text assert 'name="close_futures_position"' not in text assert 'name="set_leverage_and_margin"' not in text assert '"run_futures_cycle"' in text or "'run_futures_cycle'" in text assert "get_market_context" in text # --------------------------------------------------------------------------- # P0-T04: global execution kill switch (GAP-001) # --------------------------------------------------------------------------- def test_p0_t04_kill_switch_defaults_disabled(monkeypatch): """Unset HERMES_EXECUTION_ENABLED must mean execution is disabled.""" monkeypatch.delenv(fx.EXECUTION_ENABLED_ENV, raising=False) assert fx.is_execution_enabled() is False def test_p0_t04_kill_switch_blocks_new_entry(monkeypatch): monkeypatch.delenv(fx.EXECUTION_ENABLED_ENV, raising=False) with pytest.raises(fx.TradingError, match="kill switch"): asyncio.run(fx.execute_futures_position( symbol="BTCUSDT", side="long", leverage=5, size=1, stop_loss=95, take_profit=110, orderbook={"asks": [[100, 1000]], "bids": [[99, 1000]]}, )) def test_p0_t04_kill_switch_blocks_leverage(monkeypatch): monkeypatch.delenv(fx.EXECUTION_ENABLED_ENV, raising=False) with pytest.raises(fx.TradingError, match="kill switch"): asyncio.run(fx.set_leverage_and_margin("BTCUSDT", 5, margin_type="isolated")) def test_p0_t04_kill_switch_allows_close_only(monkeypatch): """Close remains available for emergency reduce-risk when switch is off.""" monkeypatch.setenv(fx.EXECUTION_ENABLED_ENV, "true") assert asyncio.run(_seed_protected_position())["status"] == "protected" # now disable kill switch and still allow close monkeypatch.delenv(fx.EXECUTION_ENABLED_ENV, raising=False) assert fx.is_execution_enabled() is False closed = asyncio.run(fx.close_futures_position("BTCUSDT", reason="emergency")) assert closed["status"] == "closed" def test_p0_t04_kill_switch_enabled_does_not_restore_raw_entry(monkeypatch): monkeypatch.setenv(fx.EXECUTION_ENABLED_ENV, "true") book = {"asks": [[100, 1000]], "bids": [[99, 1000]]} with pytest.raises(fx.TradingError, match="direct raw futures execution is disabled"): asyncio.run(fx.execute_futures_position( symbol="BTCUSDT", side="long", leverage=5, size=1, stop_loss=95, take_profit=110, orderbook=book, ))