trader-mcp/tests/test_engine.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, strategy_engine
from src.trader_mcp import server as trader_server
from src.trader_mcp.server import app, apply_control_decision


STRATEGY_CODE = '''
from src.trader_mcp.strategy_sdk import Strategy

class Strategy(Strategy):
    CONFIG_SCHEMA = {"label": {"type": "string", "default": "hello world"}}

    def init(self):
        return {"started": True, "config_copy": dict(self.config)}

    def on_tick(self, tick):
        self.state["ticks"] = self.state.get("ticks", 0) + 1
        return self.state["ticks"]

    def render(self):
        return {"widgets": [{"type": "metric", "label": "ticks", "value": self.state.get("ticks", 0)}]}
'''

COUNTER_STRATEGY_CODE = '''
from src.trader_mcp.strategy_sdk import Strategy

class Strategy(Strategy):
    def init(self):
        return {"counter": 0}

    def on_tick(self, tick):
        self.state["counter"] += 1
        return self.state["counter"]

    def render(self):
        return {"widgets": [{"type": "metric", "label": "ticks", "value": self.state["counter"]}]}
'''


def test_mode_off_does_not_instantiate_and_active_does(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 / "demo.py").write_text(STRATEGY_CODE)

        strategy_store.add_strategy_instance(id="s1", strategy_type="demo", account_id="acct-1", client_id="cid-1", mode="off", config={"x": 1})
        result = strategy_engine.reconcile_all()
        assert result["running"] == []
        assert strategy_engine.get_running_strategy("s1") is None

        strategy_store.update_strategy_mode("s1", "active")
        result = strategy_engine.reconcile_all()
        assert "s1" in result["running"]
        runtime = strategy_engine.get_running_strategy("s1")
        assert runtime is not None
        assert runtime.instance.state["started"] is True
        assert runtime.instance.context.account_id == "acct-1"
        assert runtime.instance.context.client_id == "cid-1"
        tick_result = strategy_engine.tick_strategy("s1", {"price": 1})
        assert tick_result["ok"] is True
        assert tick_result["result"] == 1
        render_result = strategy_engine.render_strategy("s1")
        assert render_result["ok"] is True
        assert render_result["render"]["widgets"][0]["value"] == 1
    finally:
        strategy_store.DB_PATH = original_db
        strategy_registry.STRATEGIES_DIR = original_dir
        strategy_engine._running.clear()


def test_mode_change_route_reconciles(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 / "demo.py").write_text(STRATEGY_CODE)

        client = TestClient(app)
        client.post(
            "/strategies",
            json={"id": "s2", "strategy_type": "demo", "account_id": "acct-1", "client_id": "cid-1", "mode": "off", "config": {"x": 1}},
        )
        r = client.post("/strategies/s2/mode", json={"mode": "active"})
        assert r.status_code == 200
        assert "s2" in r.json()["running"]
    finally:
        strategy_store.DB_PATH = original_db
        strategy_registry.STRATEGIES_DIR = original_dir
        strategy_engine._running.clear()


def test_dashboard_add_strategy_synthesizes_client_and_defaults(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 / "hello_world.py").write_text(STRATEGY_CODE)

        client = TestClient(app)
        response = client.post(
            "/dashboard/strategies/add",
            data={"name": "My super Grid 0.5", "strategy_type": "hello_world", "account_id": "acct-1"},
            follow_redirects=False,
        )
        assert response.status_code in {302, 303}

        record = strategy_store.list_strategy_instances()[0]
        assert record.name == "My super Grid 0.5"
        assert record.client_id.startswith("hello_world:")
        assert record.config == {"label": "hello world"}
        assert record.mode == "off"
    finally:
        strategy_store.DB_PATH = original_db
        strategy_registry.STRATEGIES_DIR = original_dir
        strategy_engine._running.clear()


def test_runtime_pause_suppresses_tick_and_render(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 / "hello_world.py").write_text(STRATEGY_CODE)

        strategy_store.add_strategy_instance(id="s3", strategy_type="hello_world", account_id="acct-1", client_id="cid-1", mode="active", config={})
        strategy_engine.reconcile_all()

        assert strategy_engine.pause_strategy("s3")["ok"] is True
        tick_result = strategy_engine.tick_strategy("s3", {"price": 1})
        assert tick_result["paused"] is True
        assert tick_result["skipped"] is True
        render_result = strategy_engine.render_strategy("s3")
        assert render_result["paused"] is True
        assert render_result["render"] is None

        assert strategy_engine.resume_strategy("s3")["ok"] is True
        tick_result = strategy_engine.tick_strategy("s3", {"price": 1})
        assert tick_result["result"] == 1
        render_result = strategy_engine.render_strategy("s3")
        assert render_result["render"]["widgets"][0]["value"] == 1
    finally:
        strategy_store.DB_PATH = original_db
        strategy_registry.STRATEGIES_DIR = original_dir
        strategy_engine._running.clear()


def test_run_due_ticks_triggers_tick_when_due(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 / "hello_world.py").write_text(COUNTER_STRATEGY_CODE)

        strategy_store.add_strategy_instance(id="s4", strategy_type="hello_world", account_id="acct-1", client_id="cid-1", mode="active", config={})
        strategy_engine.reconcile_all()
        runtime = strategy_engine.get_running_strategy("s4")
        assert runtime is not None
        runtime.next_tick_at = 0.0

        result = strategy_engine.run_due_ticks(now=1.0)
        assert "s4" in result["ticked"]
        assert strategy_engine.get_running_strategy("s4").instance.state["counter"] == 1
    finally:
        strategy_store.DB_PATH = original_db
        strategy_registry.STRATEGIES_DIR = original_dir
        strategy_engine._running.clear()


def test_dashboard_pause_toggle(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 / "hello_world.py").write_text(STRATEGY_CODE)

        client = TestClient(app)
        client.post(
            "/dashboard/strategies/add",
            data={"name": "Pause test", "strategy_type": "hello_world", "account_id": "acct-1"},
            follow_redirects=False,
        )
        strategy_id = strategy_store.list_strategy_instances()[0].id

        strategy_store.update_strategy_mode(strategy_id, "active")
        strategy_engine.reconcile_instance(strategy_id)
        assert strategy_engine.get_running_strategy(strategy_id).paused is False

        client.post(f"/dashboard/strategies/{strategy_id}/pause", follow_redirects=False)
        assert strategy_engine.get_running_strategy(strategy_id).paused is True

        client.post(f"/dashboard/strategies/{strategy_id}/pause", follow_redirects=False)
        assert strategy_engine.get_running_strategy(strategy_id).paused is False
    finally:
        strategy_store.DB_PATH = original_db
        strategy_registry.STRATEGIES_DIR = original_dir
        strategy_engine._running.clear()


def test_dashboard_detail_panel_shows_render_and_saves_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 / "hello_world.py").write_text(STRATEGY_CODE)

        client = TestClient(app)
        client.post(
            "/dashboard/strategies/add",
            data={"name": "Render test", "strategy_type": "hello_world", "account_id": "acct-1"},
            follow_redirects=False,
        )
        strategy_id = strategy_store.list_strategy_instances()[0].id
        strategy_store.update_strategy_mode(strategy_id, "active")
        strategy_engine.reconcile_instance(strategy_id)

        page = client.get("/dashboard/")
        assert page.status_code == 200
        render_response = client.get(f"/dashboard/strategies/{strategy_id}/render")
        assert render_response.status_code == 200
        assert render_response.json()["render"]["widgets"][0]["label"] == "ticks"

        client.post(
            f"/dashboard/strategies/{strategy_id}/config",
            data={"config_json": '{"label": "changed label"}'},
            follow_redirects=False,
        )
        record = strategy_store.get_strategy_instance(strategy_id)
        assert record is not None
        assert record.config == {"label": "changed label"}
    finally:
        strategy_store.DB_PATH = original_db
        strategy_registry.STRATEGIES_DIR = original_dir
        strategy_engine._running.clear()


def test_apply_control_decision_switches_active_strategy_and_records_audit(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 / "hello_world.py").write_text(STRATEGY_CODE)
        (strategy_registry.STRATEGIES_DIR / "demo.py").write_text(STRATEGY_CODE)

        strategy_store.add_strategy_instance(id="old", strategy_type="hello_world", account_id="acct-1", client_id="cid-old", mode="active", market_symbol="xrpusd", base_currency="XRP", counter_currency="USD", config={})
        strategy_store.add_strategy_instance(id="new", strategy_type="demo", account_id="acct-1", client_id="cid-new", mode="off", market_symbol="xrpusd", base_currency="XRP", counter_currency="USD", config={})
        strategy_engine.reconcile_all()

        result = apply_control_decision(
            {
                "decision_id": "dec-switch-1",
                "concern_id": "acct-1:xrpusd",
                "account_id": "acct-1",
                "market_symbol": "xrpusd",
                "action": "switch",
                "target_strategy_id": "new",
                "expected_active_strategy_id": "old",
                "reason": "trend fits better",
                "confidence": 0.9,
            }
        )
        assert result["ok"] is True
        assert result["status"] == "applied"
        assert strategy_store.get_strategy_instance("old").mode == "off"
        assert strategy_store.get_strategy_instance("new").mode == "active"
        audit = strategy_store.get_control_action_by_decision_id("dec-switch-1")
        assert audit is not None
        assert audit.status == "applied"
        assert audit.result["to_strategy_id"] == "new"
    finally:
        strategy_store.DB_PATH = original_db
        strategy_registry.STRATEGIES_DIR = original_dir
        strategy_engine._running.clear()


def test_apply_control_decision_is_idempotent_by_decision_id(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 / "hello_world.py").write_text(STRATEGY_CODE)
        (strategy_registry.STRATEGIES_DIR / "demo.py").write_text(STRATEGY_CODE)

        strategy_store.add_strategy_instance(id="old", strategy_type="hello_world", account_id="acct-1", client_id="cid-old", mode="active", market_symbol="xrpusd", base_currency="XRP", counter_currency="USD", config={})
        strategy_store.add_strategy_instance(id="new", strategy_type="demo", account_id="acct-1", client_id="cid-new", mode="off", market_symbol="xrpusd", base_currency="XRP", counter_currency="USD", config={})
        strategy_engine.reconcile_all()

        payload = {
            "decision_id": "dec-switch-2",
            "concern_id": "acct-1:xrpusd",
            "account_id": "acct-1",
            "market_symbol": "xrpusd",
            "action": "switch",
            "target_strategy_id": "new",
            "expected_active_strategy_id": "old",
            "reason": "trend fits better",
            "confidence": 0.9,
        }
        first = apply_control_decision(payload)
        second = apply_control_decision(payload)
        assert first == second
    finally:
        strategy_store.DB_PATH = original_db
        strategy_registry.STRATEGIES_DIR = original_dir
        strategy_engine._running.clear()


def test_apply_control_decision_rejects_expected_active_mismatch(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 / "hello_world.py").write_text(STRATEGY_CODE)
        (strategy_registry.STRATEGIES_DIR / "demo.py").write_text(STRATEGY_CODE)

        strategy_store.add_strategy_instance(id="old", strategy_type="hello_world", account_id="acct-1", client_id="cid-old", mode="active", market_symbol="xrpusd", base_currency="XRP", counter_currency="USD", config={})
        strategy_store.add_strategy_instance(id="new", strategy_type="demo", account_id="acct-1", client_id="cid-new", mode="off", market_symbol="xrpusd", base_currency="XRP", counter_currency="USD", config={})
        strategy_engine.reconcile_all()

        result = apply_control_decision(
            {
                "decision_id": "dec-switch-3",
                "concern_id": "acct-1:xrpusd",
                "account_id": "acct-1",
                "market_symbol": "xrpusd",
                "action": "switch",
                "target_strategy_id": "new",
                "expected_active_strategy_id": "something-else",
                "reason": "trend fits better",
                "confidence": 0.9,
            }
        )
        assert result["ok"] is False
        assert result["status"] == "rejected"
        assert "expected active strategy mismatch" in result["errors"]
    finally:
        strategy_store.DB_PATH = original_db
        strategy_registry.STRATEGIES_DIR = original_dir
        strategy_engine._running.clear()


def test_apply_control_decision_switches_even_when_current_is_degraded(tmp_path, monkeypatch):
    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 / "hello_world.py").write_text(STRATEGY_CODE)
        (strategy_registry.STRATEGIES_DIR / "demo.py").write_text(STRATEGY_CODE)

        strategy_store.add_strategy_instance(id="old", strategy_type="hello_world", account_id="acct-1", client_id="cid-old", mode="active", market_symbol="xrpusd", base_currency="XRP", counter_currency="USD", config={})
        strategy_store.add_strategy_instance(id="new", strategy_type="demo", account_id="acct-1", client_id="cid-new", mode="off", market_symbol="xrpusd", base_currency="XRP", counter_currency="USD", config={})
        strategy_engine.reconcile_all()

        monkeypatch.setattr(trader_server, "_is_degraded", lambda instance_id: True)

        result = apply_control_decision(
            {
                "decision_id": "dec-switch-degraded-1",
                "concern_id": "acct-1:xrpusd",
                "account_id": "acct-1",
                "market_symbol": "xrpusd",
                "action": "switch",
                "target_strategy_id": "new",
                "expected_active_strategy_id": "old",
                "reason": "trend fits better",
                "confidence": 0.9,
            }
        )
        assert result["ok"] is True
        assert result["status"] == "applied"
        assert result["validation"]["current_degraded"] is True
        assert result["result"].get("current_degraded") is True
        assert result["result"].get("warning")
        assert strategy_store.get_strategy_instance("old").mode == "off"
        assert strategy_store.get_strategy_instance("new").mode == "active"
    finally:
        strategy_store.DB_PATH = original_db
        strategy_registry.STRATEGIES_DIR = original_dir
        strategy_engine._running.clear()