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 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 / "stop_loss_trader.py").write_text((Path(__file__).resolve().parents[1] / "strategies" / "stop_loss_trader.py").read_text()) grid_defaults = strategy_registry.get_strategy_default_config("grid_trader") stop_defaults = strategy_registry.get_strategy_default_config("stop_loss_trader") assert grid_defaults["trade_sides"] == "both" assert grid_defaults["trend_guard_reversal_max"] == 0.25 assert stop_defaults["regime_timeframes"] == ["1d", "4h", "1h", "15m"] assert stop_defaults["trend_enter_threshold"] == 0.7 assert stop_defaults["trend_exit_threshold"] == 0.45 finally: strategy_store.DB_PATH = original_db strategy_registry.STRATEGIES_DIR = original_dir def test_grid_top_up_uses_missing_levels_budget(): class FakeContext: base_currency = "XRP" counter_currency = "USD" market_symbol = "xrpusd" minimum_order_value = 10.0 mode = "active" def __init__(self): 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, inventory_cap_pct=0.0, 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 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["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"] strategy._top_up_missing_levels(strategy.state["center_price"], strategy.state["orders"]) assert len(ctx.placed_orders) == 1 assert ctx.placed_orders[0]["side"] == "buy" assert float(ctx.placed_orders[0]["amount"]) > 7.57