trader-mcp/tests/test_strategies.py

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