from __future__ import annotations from pathlib import Path from tempfile import TemporaryDirectory from fastapi.testclient import TestClient from src.trader_mcp import strategy_registry, strategy_store from src.trader_mcp.server import app from src.trader_mcp.strategy_context import StrategyContext from src.trader_mcp.strategy_sizing import cap_amount_to_balance_target from src.trader_mcp.strategy_sdk import Strategy as BaseStrategy from strategies.exposure_protector import Strategy as ExposureStrategy from strategies.dumb_trader import Strategy as DumbStrategy from strategies.grid_trader import Strategy as GridStrategy STRATEGY_CODE = ''' from src.trader_mcp.strategy_sdk import Strategy class Strategy(Strategy): def init(self): return {"started": True, "config_copy": dict(self.config)} ''' def test_strategies_endpoints_roundtrip(): with TemporaryDirectory() as tmpdir: strategy_store.DB_PATH = Path(tmpdir) / "trader_mcp.sqlite3" from src.trader_mcp import strategy_registry strategy_registry.STRATEGIES_DIR = Path(tmpdir) / "strategies" strategy_registry.STRATEGIES_DIR.mkdir() (strategy_registry.STRATEGIES_DIR / "demo.py").write_text(STRATEGY_CODE) client = TestClient(app) r = client.get("/strategies") assert r.status_code == 200 body = r.json() assert "available" in body assert "configured" in body r = client.post( "/strategies", json={ "id": "demo-1", "strategy_type": "demo", "account_id": "acct-1", "client_id": "strategy:test", "mode": "observe", "config": {"risk": 0.01}, }, ) assert r.status_code == 200 assert r.json()["id"] == "demo-1" r = client.get("/strategies") assert any(item["id"] == "demo-1" for item in r.json()["configured"]) r = client.delete("/strategies/demo-1") assert r.status_code == 200 assert r.json()["ok"] is True def test_strategy_context_binds_identity(monkeypatch): calls = {} def fake_place_order(arguments): calls["place_order"] = arguments return {"ok": True} def fake_open_orders(account_id, client_id=None): calls["open_orders"] = {"account_id": account_id, "client_id": client_id} return {"ok": True} def fake_cancel_all(account_id, client_id=None): calls["cancel_all"] = {"account_id": account_id, "client_id": client_id} return {"ok": True} monkeypatch.setattr("src.trader_mcp.strategy_context.place_order", fake_place_order) monkeypatch.setattr("src.trader_mcp.strategy_context.list_open_orders", fake_open_orders) monkeypatch.setattr("src.trader_mcp.strategy_context.cancel_all_orders", fake_cancel_all) ctx = StrategyContext(id="inst-1", account_id="acct-1", client_id="client-1", mode="active") ctx.place_order(side="sell", market="xrpusd", order_type="limit", amount="10", price="2") ctx.get_open_orders() ctx.cancel_all_orders() assert calls["place_order"]["account_id"] == "acct-1" assert calls["place_order"]["client_id"] == "client-1" assert calls["open_orders"] == {"account_id": "acct-1", "client_id": "client-1"} assert calls["cancel_all"] == {"account_id": "acct-1", "client_id": "client-1"} def test_stop_loss_strategy_loads_with_aligned_regime_config(tmp_path): original_db = strategy_store.DB_PATH original_dir = strategy_registry.STRATEGIES_DIR try: strategy_store.DB_PATH = tmp_path / "trader_mcp.sqlite3" strategy_registry.STRATEGIES_DIR = tmp_path / "strategies" strategy_registry.STRATEGIES_DIR.mkdir() (strategy_registry.STRATEGIES_DIR / "grid_trader.py").write_text((Path(__file__).resolve().parents[1] / "strategies" / "grid_trader.py").read_text()) (strategy_registry.STRATEGIES_DIR / "exposure_protector.py").write_text((Path(__file__).resolve().parents[1] / "strategies" / "exposure_protector.py").read_text()) grid_defaults = strategy_registry.get_strategy_default_config("grid_trader") stop_defaults = strategy_registry.get_strategy_default_config("exposure_protector") assert grid_defaults["trade_sides"] == "both" assert grid_defaults["grid_step_pct"] == 0.012 assert stop_defaults["trail_distance_pct"] == 0.03 assert stop_defaults["rebalance_target_ratio"] == 0.5 assert stop_defaults["min_rebalance_seconds"] == 180 assert stop_defaults["min_price_move_pct"] == 0.005 finally: strategy_store.DB_PATH = original_db strategy_registry.STRATEGIES_DIR = original_dir def test_grid_supervision_reports_factual_capacity_not_handoff_commands(): class FakeContext: account_id = "acct-1" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" mode = "active" strategy = GridStrategy(FakeContext(), {}) strategy.state.update({ "last_price": 1.45, "base_available": 50.0, "counter_available": 38.7, "regimes": {"1h": {"trend": {"state": "bull"}}}, }) supervision = strategy._supervision() assert supervision["inventory_pressure"] == "base_heavy" assert supervision["capacity_available"] is False assert supervision["side_capacity"] == {"buy": True, "sell": True} strategy.state.update({ "base_available": 88.0, "counter_available": 4.0, }) supervision = strategy._supervision() assert supervision["inventory_pressure"] == "base_side_depleted" assert supervision["side_capacity"] == {"buy": True, "sell": False} def test_grid_supervision_exposes_adverse_side_open_orders(): class FakeContext: account_id = "acct-1" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" mode = "active" strategy = GridStrategy(FakeContext(), {}) strategy.state.update({ "last_price": 1.60, "center_price": 1.45, "orders": [ {"side": "sell", "price": "1.62", "amount": "10", "status": "open"}, {"side": "sell", "price": "1.66", "amount": "5", "status": "open"}, {"side": "buy", "price": "1.38", "amount": "7", "status": "open"}, ], }) supervision = strategy._supervision() assert supervision["market_bias"] == "bullish" assert supervision["adverse_side"] == "sell" assert supervision["adverse_side_open_order_count"] == 2 assert supervision["adverse_side_open_order_notional_quote"] > 0 assert "sell ladder exposed" in " ".join(supervision["concerns"]) def test_dumb_trader_and_protector_supervision_reports_facts_only(): class FakeContext: account_id = "acct-1" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" mode = "active" trend = DumbStrategy(FakeContext(), {"trade_side": "buy", "order_notional_quote": 1.0}) trend.state.update({"last_price": 1.45, "base_available": 20.0, "counter_available": 20.0, "last_order_at": 0.0}) trend_supervision = trend._supervision() assert trend_supervision["trade_side"] == "buy" assert trend_supervision["capacity_available"] is True assert trend_supervision["entry_offset_pct"] == 0.003 assert trend_supervision["chasing_risk"] in {"low", "moderate", "elevated"} assert "switch_readiness" not in trend_supervision assert "desired_companion" not in trend_supervision protector = ExposureStrategy(FakeContext(), {}) protector.state.update({"last_price": 1.45, "base_available": 40.0, "counter_available": 10.0}) protector_supervision = protector._supervision() assert protector_supervision["rebalance_needed"] is True assert protector_supervision["repair_progress"] <= 1.0 assert "switch_readiness" not in protector_supervision assert "desired_companion" not in protector_supervision def test_exposure_protector_holds_inside_hysteresis_band(monkeypatch): class FakeContext: account_id = "acct-1" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" mode = "active" def __init__(self): self.placed_orders = [] def get_account_info(self): return {"balances": [{"asset_code": "XRP", "available": 9.2}, {"asset_code": "USD", "available": 10.0}]} def get_price(self, market): return {"price": 1.0} def get_fee_rates(self, market): return {"maker": 0.0, "taker": 0.004} def place_order(self, **kwargs): self.placed_orders.append(kwargs) return {"status": "ok", "id": f"oid-{len(self.placed_orders)}"} ctx = FakeContext() strategy = ExposureStrategy(ctx, {"rebalance_target_ratio": 0.5, "rebalance_step_ratio": 0.15, "balance_tolerance": 0.05, "cooldown_ticks": 0, "min_rebalance_seconds": 0, "trail_distance_pct": 0.03}) strategy.state["last_rebalance_side"] = "sell" strategy.state["last_order_at"] = 0 monkeypatch.setattr(strategy, "_refresh_regimes", lambda: None) result = strategy.on_tick({}) assert result["action"] == "hold" assert result["reason"] == "within rebalance hysteresis" assert ctx.placed_orders == [] def test_grid_apply_policy_keeps_explicit_grid_levels(): class FakeContext: account_id = "acct-1" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" mode = "active" strategy = GridStrategy(FakeContext(), {"grid_levels": 5, "policy": {"risk_posture": "normal"}}) strategy.apply_policy() assert strategy.config["grid_levels"] == 5 assert strategy.state["policy_derived"]["grid_levels"] == 5 def test_grid_seed_keeps_other_side_when_one_side_fails(monkeypatch): class FakeContext: base_currency = "XRP" counter_currency = "USD" market_symbol = "xrpusd" minimum_order_value = 10.0 mode = "active" def __init__(self): self.attempts = [] self.buy_attempts = 0 self.sell_attempts = 0 def get_fee_rates(self, market): return {"maker": 0.0, "taker": 0.0} def suggest_order_amount(self, **kwargs): return 10.0 def place_order(self, **kwargs): self.attempts.append(kwargs) if kwargs["side"] == "buy": self.buy_attempts += 1 if self.buy_attempts == 3: raise RuntimeError("insufficient USD") elif kwargs["side"] == "sell": self.sell_attempts += 1 return {"status": "ok", "id": f"{kwargs['side']}-{len(self.attempts)}"} ctx = FakeContext() strategy = GridStrategy(ctx, {"grid_levels": 5, "order_call_delay_ms": 0, "enable_trend_guard": False, "fee_rate": 0.0}) strategy.state["center_price"] = 100.0 monkeypatch.setattr(strategy, "_supported_levels", lambda side, center, min_notional: 5) monkeypatch.setattr(strategy, "_refresh_balance_snapshot", lambda: None) strategy._place_grid(100.0) orders = strategy.state["orders"] assert ctx.buy_attempts == 5 assert ctx.sell_attempts == 5 assert len([o for o in orders if o["side"] == "buy"]) == 4 assert len([o for o in orders if o["side"] == "sell"]) == 5 assert any("partial success" in line for line in (strategy.state.get("debug_log") or [])) or strategy.state.get("last_error") == "insufficient USD" def test_grid_skips_rebuild_when_balance_refresh_fails(monkeypatch): class FakeContext: base_currency = "XRP" counter_currency = "USD" market_symbol = "xrpusd" minimum_order_value = 10.0 mode = "active" def __init__(self): self.cancelled_all = 0 self.placed_orders = [] def get_fee_rates(self, market): return {"maker": 0.0, "taker": 0.004} def get_account_info(self): raise RuntimeError("Bitstamp auth breaker active, retry later") def cancel_all_orders(self): self.cancelled_all += 1 return {"ok": True} def suggest_order_amount(self, **kwargs): return 10.0 def place_order(self, **kwargs): self.placed_orders.append(kwargs) return {"status": "ok", "id": f"oid-{len(self.placed_orders)}"} ctx = FakeContext() strategy = GridStrategy(ctx, {"grid_levels": 5, "order_call_delay_ms": 0, "enable_trend_guard": False, "fee_rate": 0.004}) strategy.state["center_price"] = 1.4397 strategy.state["seeded"] = True strategy.state["orders"] = [{"side": "buy", "price": 1.43, "amount": 10.0, "id": "o1"}] strategy.state["order_ids"] = ["o1"] monkeypatch.setattr(strategy, "_sync_open_orders_state", lambda: [{"side": "buy", "price": 1.43, "amount": 10.0, "id": "o1"}]) monkeypatch.setattr(strategy, "_price", lambda: 1.4397) monkeypatch.setattr(strategy, "_refresh_regimes", lambda: None) result = strategy.on_tick({}) assert result["action"] == "hold" assert result["reason"] == "balance refresh unavailable" assert ctx.cancelled_all == 0 assert ctx.placed_orders == [] def test_grid_skips_shape_rebuild_when_balance_reads_turn_inconclusive(monkeypatch): class FakeContext: base_currency = "XRP" counter_currency = "USD" market_symbol = "xrpusd" minimum_order_value = 10.0 mode = "active" def __init__(self): self.cancelled_all = 0 self.placed_orders = [] self.calls = 0 def get_fee_rates(self, market): return {"maker": 0.0, "taker": 0.004} def get_account_info(self): self.calls += 1 if self.calls == 1: return { "balances": [ {"asset_code": "USD", "available": 41.29}, {"asset_code": "XRP", "available": 9.98954}, ] } raise RuntimeError("Bitstamp auth breaker active, retry later") def cancel_all_orders(self): self.cancelled_all += 1 return {"ok": True} def suggest_order_amount(self, **kwargs): return 10.0 def place_order(self, **kwargs): self.placed_orders.append(kwargs) return {"status": "ok", "id": f"oid-{len(self.placed_orders)}"} ctx = FakeContext() strategy = GridStrategy(ctx, {"grid_levels": 2, "order_call_delay_ms": 0, "enable_trend_guard": False, "fee_rate": 0.004}) strategy.state["center_price"] = 1.3285 strategy.state["seeded"] = True strategy.state["orders"] = [ {"side": "buy", "price": 1.3243993, "amount": 7.63, "id": "existing-buy"}, {"side": "sell", "price": 1.3326007, "amount": 9.0, "id": "sell-1"}, {"side": "sell", "price": 1.3367011, "amount": 9.0, "id": "sell-2"}, ] strategy.state["order_ids"] = ["existing-buy", "sell-1", "sell-2"] def fake_sync_open_orders_state(): live = [{"side": "sell", "price": 1.3326007, "amount": 9.0, "id": "sell-1"}] strategy.state["orders"] = live strategy.state["order_ids"] = ["sell-1"] strategy.state["open_order_count"] = 1 return live monkeypatch.setattr(strategy, "_sync_open_orders_state", fake_sync_open_orders_state) monkeypatch.setattr(strategy, "_price", lambda: 1.3285) monkeypatch.setattr(strategy, "_refresh_regimes", lambda: None) monkeypatch.setattr(strategy, "_grid_refresh_paused", lambda: False) monkeypatch.setattr(strategy, "_recenter_threshold_pct", lambda: 0.5) result = strategy.on_tick({}) assert result["action"] == "hold" assert ctx.cancelled_all == 0 assert ctx.placed_orders == [] def test_grid_missing_order_triggers_full_rebuild(monkeypatch): class FakeContext: base_currency = "XRP" counter_currency = "USD" market_symbol = "xrpusd" minimum_order_value = 10.0 mode = "active" def __init__(self): self.cancelled_all = 0 self.placed_orders = [] def get_fee_rates(self, market): return {"maker": 0.0, "taker": 0.004} def get_account_info(self): return { "balances": [ {"asset_code": "USD", "available": 13.55}, {"asset_code": "XRP", "available": 22.0103}, ] } def suggest_order_amount( self, *, side, price, levels, min_notional, fee_rate, max_notional_per_order=0.0, dust_collect=False, order_size=0.0, safety=0.995, ): if side == "buy": quote_available = 13.55 spendable_quote = quote_available * safety quote_cap = min(spendable_quote, max_notional_per_order) if max_notional_per_order > 0 else spendable_quote if quote_cap < min_notional * (1 + fee_rate): return 0.0 return quote_cap / (price * (1 + fee_rate)) return 0.0 def cancel_all_orders(self): self.cancelled_all += 1 return {"ok": True} def place_order(self, **kwargs): self.placed_orders.append(kwargs) return {"status": "ok", "id": f"oid-{len(self.placed_orders)}"} ctx = FakeContext() strategy = GridStrategy( ctx, { "grid_levels": 2, "grid_step_pct": 0.0062, "grid_step_min_pct": 0.0033, "grid_step_max_pct": 0.012, "max_notional_per_order": 12, "order_call_delay_ms": 0, "trade_sides": "both", "debug_orders": True, "dust_collect": True, "enable_trend_guard": False, "fee_rate": 0.004, }, ) strategy.state["center_price"] = 1.3285 strategy.state["seeded"] = True strategy.state["base_available"] = 22.0103 strategy.state["counter_available"] = 13.55 strategy.state["orders"] = [ {"side": "buy", "price": 1.3243993, "amount": 7.63, "id": "existing-buy"}, {"side": "sell", "price": 1.3326007, "amount": 9.0, "id": "sell-1"}, {"side": "sell", "price": 1.3367011, "amount": 9.0, "id": "sell-2"}, ] strategy.state["order_ids"] = ["existing-buy", "sell-1", "sell-2"] def fake_sync_open_orders_state(): live = [{"side": "sell", "price": 1.3326007, "amount": 9.0, "id": "sell-1"}] strategy.state["orders"] = live strategy.state["order_ids"] = ["sell-1"] strategy.state["open_order_count"] = 1 return live monkeypatch.setattr(strategy, "_sync_open_orders_state", fake_sync_open_orders_state) monkeypatch.setattr(strategy, "_price", lambda: 1.3285) monkeypatch.setattr(strategy, "_refresh_regimes", lambda: None) monkeypatch.setattr(strategy, "_grid_refresh_paused", lambda: False) monkeypatch.setattr(strategy, "_recenter_threshold_pct", lambda: 0.5) result = strategy.on_tick({}) assert result["action"] in {"seed", "reseed"} assert ctx.cancelled_all == 1 assert len(ctx.placed_orders) > 0 assert strategy.state["last_action"] == "reseeded" def test_grid_side_imbalance_triggers_full_rebuild(monkeypatch): class FakeContext: base_currency = "XRP" counter_currency = "USD" market_symbol = "xrpusd" minimum_order_value = 10.0 mode = "active" def __init__(self): self.cancelled_all = 0 self.placed_orders = [] def get_fee_rates(self, market): return {"maker": 0.0, "taker": 0.004} def get_account_info(self): return {"balances": [{"asset_code": "USD", "available": 41.29}, {"asset_code": "XRP", "available": 9.98954}]} def cancel_all_orders(self): self.cancelled_all += 1 return {"ok": True} def suggest_order_amount(self, **kwargs): return 10.0 def place_order(self, **kwargs): self.placed_orders.append(kwargs) return {"status": "ok", "id": f"oid-{len(self.placed_orders)}"} ctx = FakeContext() strategy = GridStrategy(ctx, {"grid_levels": 2, "order_call_delay_ms": 0, "enable_trend_guard": False, "fee_rate": 0.004}) strategy.state["center_price"] = 1.3907 strategy.state["seeded"] = True strategy.state["orders"] = [{"side": "buy", "price": 1.3800, "amount": 10.0, "id": "o1"} for _ in range(5)] strategy.state["order_ids"] = [f"o{i}" for i in range(5)] def fake_sync_open_orders_state(): live = [{"side": "buy", "price": 1.3800, "amount": 10.0, "id": f"o{i}"} for i in range(5)] strategy.state["orders"] = live strategy.state["order_ids"] = [f"o{i}" for i in range(5)] strategy.state["open_order_count"] = 5 return live monkeypatch.setattr(strategy, "_sync_open_orders_state", fake_sync_open_orders_state) monkeypatch.setattr(strategy, "_refresh_balance_snapshot", lambda: True) monkeypatch.setattr(strategy, "_price", lambda: 1.3915) monkeypatch.setattr(strategy, "_refresh_regimes", lambda: None) monkeypatch.setattr(strategy, "_grid_refresh_paused", lambda: False) monkeypatch.setattr(strategy, "_recenter_threshold_pct", lambda: 0.5) result = strategy.on_tick({}) assert result["action"] in {"seed", "reseed"} assert ctx.cancelled_all == 1 assert len(ctx.placed_orders) > 0 def test_grid_recenters_exactly_on_live_price(): class FakeContext: base_currency = "XRP" counter_currency = "USD" market_symbol = "xrpusd" minimum_order_value = 10.0 mode = "active" def cancel_all_orders(self): return {"ok": True} def get_fee_rates(self, market): return {"maker": 0.0, "taker": 0.0} def suggest_order_amount(self, **kwargs): return 0.1 def place_order(self, **kwargs): return {"status": "ok", "id": "oid-1"} strategy = GridStrategy(FakeContext(), {}) strategy.state["center_price"] = 100.0 strategy._recenter_and_rebuild_from_price(160.0, "test recenter") assert strategy.state["center_price"] == 160.0 def test_grid_stop_cancels_all_open_orders(): class FakeContext: base_currency = "XRP" counter_currency = "USD" market_symbol = "xrpusd" minimum_order_value = 10.0 mode = "active" def __init__(self): self.cancelled = False def cancel_all_orders(self): self.cancelled = True return {"ok": True} def get_fee_rates(self, market): return {"maker": 0.0, "taker": 0.0} strategy = GridStrategy(FakeContext(), {}) strategy.state["orders"] = [{"id": "o1"}] strategy.state["order_ids"] = ["o1"] strategy.state["open_order_count"] = 1 strategy.on_stop() assert strategy.context.cancelled is True assert strategy.state["open_order_count"] == 0 assert strategy.state["last_action"] == "stopped" def test_base_strategy_report_uses_context_snapshot(): class FakeContext: id = "s-1" account_id = "acct-1" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" mode = "active" def get_strategy_snapshot(self): return { "identity": {"strategy_id": "s-1", "strategy_name": "Demo", "account_id": "acct-1", "market": "xrpusd", "base_currency": "XRP", "quote_currency": "USD"}, "control": {"enabled_state": "on", "mode": "active"}, "position": {"balances": [{"asset_code": "XRP", "available": 1.0}]}, "orders": {"open_orders": [{"id": "o1"}]}, "execution": {"execution_quality": "good"}, } class DemoStrategy(BaseStrategy): LABEL = "Demo" report = DemoStrategy(FakeContext(), {}).report() assert report["identity"]["strategy_id"] == "s-1" assert report["control"]["mode"] == "active" assert report["position"]["open_orders"][0]["id"] == "o1" def test_dumb_trader_uses_policy_and_reports_fit(): class FakeContext: id = "s-2" account_id = "acct-2" client_id = "cid-2" mode = "active" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" def get_price(self, symbol): return {"price": 1.2} def place_order(self, **kwargs): return {"ok": True, "order": kwargs} def get_strategy_snapshot(self): return {"identity": {}, "control": {}, "position": {}, "orders": {}, "execution": {}} strat = DumbStrategy(FakeContext(), {"trade_side": "buy", "order_notional_quote": 1.5}) strat.apply_policy() report = strat.report() assert report["fit"]["risk_profile"] == "neutral" assert strat.state["policy_derived"]["order_notional_quote"] > 0 def test_dumb_trader_buys_on_configured_side_without_regime_input(): class FakeContext: id = "s-bull" account_id = "acct-1" client_id = "cid-1" mode = "active" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" def __init__(self): self.orders = [] def get_price(self, symbol): return {"price": 1.2} def place_order(self, **kwargs): self.orders.append(kwargs) return {"ok": True, "order": kwargs} def get_account_info(self): return {"balances": [{"asset_code": "USD", "available": 1000}, {"asset_code": "XRP", "available": 0}]} minimum_order_value = 10.0 def suggest_order_amount(self, **kwargs): return float(kwargs.get("quote_notional") or 0.0) / float(kwargs.get("price") or 1.0) def get_strategy_snapshot(self): return {"identity": {}, "control": {}, "position": {}, "orders": {}, "execution": {}} ctx = FakeContext() strat = DumbStrategy(ctx, {"trade_side": "buy", "order_notional_quote": 20.0}) result = strat.on_tick({}) assert result["action"] == "buy" assert ctx.orders[-1]["side"] == "buy" assert ctx.orders[-1]["amount"] == 20.0 / 1.2 assert strat.state["last_action"] == "buy_dumb" def test_dumb_trader_sells_on_configured_side_without_regime_input(): class FakeContext: id = "s-bear" account_id = "acct-1" client_id = "cid-1" mode = "active" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" def __init__(self): self.orders = [] def get_price(self, symbol): return {"price": 1.2} def place_order(self, **kwargs): self.orders.append(kwargs) return {"ok": True, "order": kwargs} def get_account_info(self): return {"balances": [{"asset_code": "USD", "available": 1000}, {"asset_code": "XRP", "available": 10}]} minimum_order_value = 10.0 def suggest_order_amount(self, **kwargs): return float(kwargs.get("quote_notional") or 0.0) / float(kwargs.get("price") or 1.0) def get_strategy_snapshot(self): return {"identity": {}, "control": {}, "position": {}, "orders": {}, "execution": {}} ctx = FakeContext() strat = DumbStrategy(ctx, {"trade_side": "sell", "order_notional_quote": 20.0}) result = strat.on_tick({}) assert result["action"] == "sell" assert ctx.orders[-1]["side"] == "sell" assert ctx.orders[-1]["amount"] == 20.0 / 1.2 assert strat.state["last_action"] == "sell_dumb" def test_dumb_trader_buy_only_ignores_bear_regime(): class FakeContext: id = "s-buy-only" account_id = "acct-1" client_id = "cid-1" mode = "active" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" def __init__(self): self.orders = [] def get_price(self, symbol): return {"price": 1.2} def get_regime(self, symbol, timeframe="1h"): return { "trend": {"state": "bear", "ema_fast": 1.17, "ema_slow": 1.2}, "momentum": {"state": "bear", "rsi": 36, "macd_histogram": -0.002}, } def place_order(self, **kwargs): self.orders.append(kwargs) return {"ok": True, "order": kwargs} def get_account_info(self): return {"balances": [{"asset_code": "USD", "available": 1000}, {"asset_code": "XRP", "available": 10}]} minimum_order_value = 10.0 def suggest_order_amount(self, **kwargs): return 10.0 def get_strategy_snapshot(self): return {"identity": {}, "control": {}, "position": {}, "orders": {}, "execution": {}} ctx = FakeContext() strat = DumbStrategy(ctx, {"trade_side": "buy", "order_notional_quote": 20.0}) result = strat.on_tick({}) assert result["action"] == "buy" assert ctx.orders[-1]["side"] == "buy" assert strat.state["last_action"] == "buy_dumb" def test_dumb_trader_sell_only_ignores_bull_regime(): class FakeContext: id = "s-sell-only" account_id = "acct-1" client_id = "cid-1" mode = "active" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" def __init__(self): self.orders = [] def get_price(self, symbol): return {"price": 1.2} def get_regime(self, symbol, timeframe="1h"): return { "trend": {"state": "bull", "ema_fast": 1.21, "ema_slow": 1.18}, "momentum": {"state": "bull", "rsi": 64, "macd_histogram": 0.002}, } def place_order(self, **kwargs): self.orders.append(kwargs) return {"ok": True, "order": kwargs} def get_account_info(self): return {"balances": [{"asset_code": "USD", "available": 1000}, {"asset_code": "XRP", "available": 0}]} minimum_order_value = 10.0 def suggest_order_amount(self, **kwargs): return 10.0 def get_strategy_snapshot(self): return {"identity": {}, "control": {}, "position": {}, "orders": {}, "execution": {}} ctx = FakeContext() strat = DumbStrategy(ctx, {"trade_side": "sell", "order_notional_quote": 20.0}) result = strat.on_tick({}) assert result["action"] == "sell" assert ctx.orders[-1]["side"] == "sell" assert strat.state["last_action"] == "sell_dumb" def test_dumb_trader_policy_does_not_override_explicit_order_notional_quote(): class FakeContext: id = "s-explicit" account_id = "acct-1" client_id = "cid-1" mode = "active" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" def get_price(self, symbol): return {"price": 1.2} def get_strategy_snapshot(self): return {"identity": {}, "control": {}, "position": {}, "orders": {}, "execution": {}} strat = DumbStrategy(FakeContext(), {"trade_side": "buy", "order_notional_quote": 10.5}) strat.apply_policy() assert strat.config["order_notional_quote"] == 10.5 assert strat.state["policy_derived"]["order_notional_quote"] == 10.5 def test_dumb_trader_passes_live_fee_rate_into_sizing_helper(): class FakeContext: id = "s-fee" account_id = "acct-1" client_id = "cid-1" mode = "active" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" minimum_order_value = 10.0 def __init__(self): self.fee_calls = [] self.suggest_calls = [] def get_price(self, symbol): return {"price": 1.2} def get_fee_rates(self, market_symbol=None): self.fee_calls.append(market_symbol) return {"maker": 0.0025, "taker": 0.004} def suggest_order_amount(self, **kwargs): self.suggest_calls.append(kwargs) return 8.0 def place_order(self, **kwargs): return {"ok": True, "order": kwargs} def get_account_info(self): return {"balances": [{"asset_code": "USD", "available": 1000}, {"asset_code": "XRP", "available": 0}]} def get_strategy_snapshot(self): return {"identity": {}, "control": {}, "position": {}, "orders": {}, "execution": {}} ctx = FakeContext() strat = DumbStrategy(ctx, {"trade_side": "buy", "order_notional_quote": 10.5, "dust_collect": True}) strat.on_tick({}) assert ctx.fee_calls == ["xrpusd"] assert ctx.suggest_calls[-1]["fee_rate"] == 0.0025 assert ctx.suggest_calls[-1]["dust_collect"] is True def test_grid_sizing_helper_receives_quote_controls_and_dust_collect(): class FakeContext: base_currency = "XRP" counter_currency = "USD" market_symbol = "xrpusd" minimum_order_value = 10.0 mode = "active" def __init__(self): self.suggest_calls = [] def get_fee_rates(self, market_symbol=None): return {"maker": 0.001, "taker": 0.004} def suggest_order_amount(self, **kwargs): self.suggest_calls.append(kwargs) return 7.0 ctx = FakeContext() strategy = GridStrategy( ctx, { "grid_levels": 3, "order_notional_quote": 11.0, "max_order_notional_quote": 12.0, "dust_collect": True, }, ) amount = strategy._suggest_amount("buy", 1.5, 3, 10.0) assert amount == 7.0 assert ctx.suggest_calls[-1]["fee_rate"] == 0.004 assert ctx.suggest_calls[-1]["quote_notional"] == 11.0 assert ctx.suggest_calls[-1]["max_notional_per_order"] == 12.0 assert ctx.suggest_calls[-1]["dust_collect"] is True assert ctx.suggest_calls[-1]["levels"] == 3 def test_dumb_trader_buy_uses_requested_notional_even_with_balance_target_configured(): class FakeContext: id = "s-buy-clamp" account_id = "acct-1" client_id = "cid-1" mode = "active" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" minimum_order_value = 0.5 def __init__(self): self.orders = [] def get_price(self, symbol): return {"price": 1.0} def get_fee_rates(self, market_symbol=None): return {"maker": 0.0, "taker": 0.0} def suggest_order_amount(self, **kwargs): return float(kwargs.get("quote_notional") or 0.0) / float(kwargs.get("price") or 1.0) def place_order(self, **kwargs): self.orders.append(kwargs) return {"ok": True, "order": kwargs} def get_account_info(self): return {"balances": [{"asset_code": "USD", "available": 6.0}, {"asset_code": "XRP", "available": 4.0}]} def get_strategy_snapshot(self): return {"identity": {}, "control": {}, "position": {}, "orders": {}, "execution": {}} ctx = FakeContext() strat = DumbStrategy(ctx, {"trade_side": "buy", "order_notional_quote": 3.0, "balance_target": 0.5}) result = strat.on_tick({}) assert result["action"] == "buy" assert ctx.orders[-1]["amount"] == 3.0 def test_dumb_trader_sell_uses_requested_notional_even_with_balance_target_configured(): class FakeContext: id = "s-sell-clamp" account_id = "acct-1" client_id = "cid-1" mode = "active" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" minimum_order_value = 0.5 def __init__(self): self.orders = [] def get_price(self, symbol): return {"price": 1.0} def get_fee_rates(self, market_symbol=None): return {"maker": 0.0, "taker": 0.0} def suggest_order_amount(self, **kwargs): return float(kwargs.get("quote_notional") or 0.0) / float(kwargs.get("price") or 1.0) def place_order(self, **kwargs): self.orders.append(kwargs) return {"ok": True, "order": kwargs} def get_account_info(self): return {"balances": [{"asset_code": "USD", "available": 3.0}, {"asset_code": "XRP", "available": 7.0}]} def get_strategy_snapshot(self): return {"identity": {}, "control": {}, "position": {}, "orders": {}, "execution": {}} ctx = FakeContext() strat = DumbStrategy(ctx, {"trade_side": "sell", "order_notional_quote": 5.0, "balance_target": 0.5}) result = strat.on_tick({}) assert result["action"] == "sell" assert ctx.orders[-1]["amount"] == 5.0 def test_dumb_trader_sell_holds_sub_minimum_order(): class FakeContext: id = "s-sell-min" account_id = "acct-1" client_id = "cid-1" mode = "active" market_symbol = "solusd" base_currency = "SOL" counter_currency = "USD" minimum_order_value = 1.0 def __init__(self): self.orders = [] def get_price(self, symbol): return {"price": 86.20062} def get_fee_rates(self, market_symbol=None): return {"maker": 0.0, "taker": 0.0} def suggest_order_amount(self, **kwargs): return 0.001 def place_order(self, **kwargs): self.orders.append(kwargs) return {"ok": True, "order": kwargs} def get_account_info(self): return {"balances": [{"asset_code": "USD", "available": 1000.0}, {"asset_code": "SOL", "available": 0.00447}]} def get_strategy_snapshot(self): return {"identity": {}, "control": {}, "position": {}, "orders": {}, "execution": {}} ctx = FakeContext() strat = DumbStrategy(ctx, {"trade_side": "sell", "order_notional_quote": 5.0, "balance_target": 1.0}) result = strat.on_tick({}) assert result["action"] == "hold" assert result["reason"] == "no usable size" assert ctx.orders == [] def test_dumb_trader_holds_when_trade_side_is_symmetrical(): class FakeContext: id = "s-target-hold" account_id = "acct-1" client_id = "cid-1" mode = "active" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" minimum_order_value = 0.5 def get_price(self, symbol): return {"price": 1.0} def get_fee_rates(self, market_symbol=None): return {"maker": 0.0, "taker": 0.0} def suggest_order_amount(self, **kwargs): return 10.0 def get_account_info(self): return {"balances": [{"asset_code": "USD", "available": 5.0}, {"asset_code": "XRP", "available": 5.0}]} def get_strategy_snapshot(self): return {"identity": {}, "control": {}, "position": {}, "orders": {}, "execution": {}} strat = DumbStrategy(FakeContext(), {"trade_side": "both", "order_notional_quote": 3.0, "balance_target": 0.5}) result = strat.on_tick({}) assert result["action"] == "hold" assert result["reason"] == "trade_side must be buy or sell" def test_cap_amount_to_balance_target_caps_sell_to_live_base(): amount = cap_amount_to_balance_target( suggested_amount=0.127226, side="sell", price=86.20062, fee_rate=0.0, balance_target=1.0, base_available=0.00447, counter_available=0.0, min_notional=0.0, ) assert amount == 0.00447 def test_cap_amount_to_balance_target_rejects_sell_below_min_notional(): amount = cap_amount_to_balance_target( suggested_amount=0.127226, side="sell", price=86.20062, fee_rate=0.0, balance_target=1.0, base_available=0.00447, counter_available=0.0, min_notional=1.0, ) assert amount == 0.0 def test_dumb_trader_ignores_balance_target_and_keeps_size(): class FakeContext: id = "s-target-open" account_id = "acct-1" client_id = "cid-1" mode = "active" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" minimum_order_value = 0.5 def __init__(self): self.orders = [] def get_price(self, symbol): return {"price": 1.0} def get_fee_rates(self, market_symbol=None): return {"maker": 0.0, "taker": 0.0} def suggest_order_amount(self, **kwargs): return float(kwargs.get("quote_notional") or 0.0) / float(kwargs.get("price") or 1.0) def place_order(self, **kwargs): self.orders.append(kwargs) return {"ok": True, "order": kwargs} def get_account_info(self): return {"balances": [{"asset_code": "USD", "available": 6.0}, {"asset_code": "XRP", "available": 4.0}]} def get_strategy_snapshot(self): return {"identity": {}, "control": {}, "position": {}, "orders": {}, "execution": {}} ctx = FakeContext() strat = DumbStrategy(ctx, {"trade_side": "buy", "order_notional_quote": 3.0, "balance_target": 0.5}) result = strat.on_tick({}) assert result["action"] == "buy" assert ctx.orders[-1]["amount"] == 3.0 def test_exposure_protector_buy_sizing_respects_fee_when_min_order_quote_is_unaffordable(): class FakeContext: account_id = "acct-1" client_id = "cid-1" mode = "active" market_symbol = "xrpusd" base_currency = "XRP" counter_currency = "USD" def get_fee_rates(self, market_symbol=None): return {"maker": 0.1, "taker": 0.2} strategy = ExposureStrategy( FakeContext(), { "rebalance_target_ratio": 0.9, "rebalance_step_ratio": 1.0, "balance_tolerance": 0.0, "min_order_notional_quote": 10.0, }, ) strategy.state["base_available"] = 0.0 strategy.state["counter_available"] = 10.0 amount = strategy._suggest_amount("buy", 1.0) assert amount == 0.0