matitos
8b6cf8130c
Adds MultiStrategyManager: it runs several strategies concurrently, each with its own TradingEngine seeded with starting_balance * allocation, so one strategy's losses cannot eat into another's capital. Weights are normalised across the run ([1,3] -> 25%/75%). Risk limits stay per strategy (each engine owns a RiskManager). New API: POST /api/bot/start-many and GET /api/bot/status. The existing single-strategy endpoints delegate to the manager, so their behaviour is unchanged. Adds manager tests (allocation, isolation, duplicate keys) and API tests.
280 lines
10 KiB
Python
280 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from starlette.websockets import WebSocketDisconnect
|
|
|
|
from app.main import create_app
|
|
from app.market.fake import FakeMarketData
|
|
from app.runtime import Runtime
|
|
|
|
|
|
def test_version_endpoint(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
assert client.get("/api/version").status_code == 200
|
|
assert "version" in client.get("/api/version").json()
|
|
|
|
|
|
def test_health_and_strategies(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
assert client.get("/api/health").json() == {"status": "ok"}
|
|
keys = {s["key"] for s in client.get("/api/strategies").json()}
|
|
assert {"sma_cross", "rsi_reversion"} <= keys
|
|
|
|
|
|
def test_start_status_stop_flow(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
status = client.get("/api/status").json()
|
|
assert status["running"] is False
|
|
|
|
started = client.post(
|
|
"/api/bot/start",
|
|
json={"strategy_key": "sma_cross", "params": {"fast": 2, "slow": 4}},
|
|
)
|
|
assert started.status_code == 200
|
|
assert started.json()["running"] is True
|
|
|
|
assert client.get("/api/positions").status_code == 200
|
|
assert client.get("/api/orders").status_code == 200
|
|
assert client.get("/api/events").status_code == 200
|
|
|
|
stopped = client.post("/api/bot/stop").json()
|
|
assert stopped["running"] is False
|
|
|
|
|
|
def test_unknown_strategy_returns_404(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
response = client.post("/api/bot/start", json={"strategy_key": "does_not_exist"})
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_backtest_endpoint(runtime: Runtime, market: FakeMarketData) -> None:
|
|
market.candles = FakeMarketData.sine_wave(200).candles
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
response = client.post(
|
|
"/api/backtest",
|
|
json={"strategy_key": "sma_cross", "params": {"fast": 5, "slow": 20}, "limit": 200},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["strategy_key"] == "sma_cross"
|
|
assert data["error"] is None
|
|
assert len(data["equity_curve"]) == 200
|
|
assert "total_return_pct" in data["metrics"]
|
|
assert data["metrics"]["num_trades"] >= 1
|
|
|
|
|
|
def test_sweep_endpoint(runtime: Runtime, market: FakeMarketData) -> None:
|
|
market.candles = FakeMarketData.sine_wave(200).candles
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
response = client.post(
|
|
"/api/sweep",
|
|
json={
|
|
"strategy_key": "sma_cross",
|
|
"grid": {"fast": [3, 5], "slow": [10, 20]},
|
|
"limit": 200,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["combinations"] == 4
|
|
assert 1 <= len(data["rows"]) <= 4
|
|
assert "total_return_pct" in data["rows"][0]["metrics"]
|
|
|
|
|
|
def test_sweep_unknown_strategy_404(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
response = client.post("/api/sweep", json={"strategy_key": "nope"})
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_backtest_unknown_strategy_404(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
response = client.post("/api/backtest", json={"strategy_key": "nope"})
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_backtest_limit_validation(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
assert (
|
|
client.post("/api/backtest", json={"strategy_key": "sma_cross", "limit": 5}).status_code
|
|
== 422
|
|
)
|
|
|
|
|
|
def test_start_many_and_multi_status(runtime: Runtime, market: FakeMarketData) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
started = client.post(
|
|
"/api/bot/start-many",
|
|
json={
|
|
"strategies": [
|
|
{"strategy_key": "sma_cross", "params": {"fast": 2, "slow": 4}, "weight": 1},
|
|
{"strategy_key": "ema_cross", "params": {"fast": 2, "slow": 4}, "weight": 3},
|
|
],
|
|
"starting_balance": 10000,
|
|
},
|
|
)
|
|
assert started.status_code == 200
|
|
body = started.json()
|
|
assert body["running"] is True
|
|
assert len(body["strategies"]) == 2
|
|
assert [s["allocation"] for s in body["strategies"]] == [0.25, 0.75]
|
|
|
|
status = client.get("/api/bot/status").json()
|
|
assert status["running"] is True
|
|
assert len(status["strategies"]) == 2
|
|
assert status["equity"] == 10000
|
|
|
|
client.post("/api/bot/stop")
|
|
|
|
|
|
def test_start_many_rejects_unknown_strategy(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
response = client.post(
|
|
"/api/bot/start-many", json={"strategies": [{"strategy_key": "nope"}]}
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_kill_endpoint(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
client.post("/api/bot/start", json={"strategy_key": "sma_cross"})
|
|
killed = client.post("/api/bot/kill", json={"reason": "test kill"}).json()
|
|
assert killed["running"] is False
|
|
assert killed["halted"] is True
|
|
assert killed["halt_reason"] == "test kill"
|
|
|
|
|
|
def test_status_exposes_risk_fields(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
status = client.get("/api/status").json()
|
|
assert {"halted", "halt_reason", "trades_today", "day_pnl", "drawdown_pct"} <= set(status)
|
|
|
|
|
|
def test_preset_crud(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
created = client.post(
|
|
"/api/presets",
|
|
json={
|
|
"name": "My SMA",
|
|
"strategy_key": "sma_cross",
|
|
"symbol": "BTC/USDT",
|
|
"timeframe": "1h",
|
|
"params": {"fast": 5, "slow": 20},
|
|
},
|
|
)
|
|
assert created.status_code == 200
|
|
preset = created.json()
|
|
assert preset["name"] == "My SMA"
|
|
|
|
listed = client.get("/api/presets").json()
|
|
assert any(p["id"] == preset["id"] for p in listed)
|
|
|
|
assert client.delete(f"/api/presets/{preset['id']}").status_code == 200
|
|
assert client.get("/api/presets").json() == []
|
|
|
|
|
|
def test_preset_unknown_strategy_404(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
response = client.post(
|
|
"/api/presets",
|
|
json={
|
|
"strategy_key": "nope",
|
|
"symbol": "BTC/USDT",
|
|
"timeframe": "1m",
|
|
"params": {},
|
|
},
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_api_token_enforced(runtime: Runtime) -> None:
|
|
runtime.settings.api_token = "secret"
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
assert client.get("/api/status").status_code == 401
|
|
ok = client.get("/api/status", headers={"X-API-Token": "secret"})
|
|
assert ok.status_code == 200
|
|
# Health stays open.
|
|
assert client.get("/api/health").status_code == 200
|
|
|
|
|
|
def test_websocket_connects_without_any_token(runtime: Runtime) -> None:
|
|
# API_TOKEN is unset by default: the API and the live feed must work with no
|
|
# token at all, even when (as nginx does) an empty token header is sent.
|
|
runtime.settings.api_token = ""
|
|
app = create_app(runtime.settings, runtime)
|
|
with (
|
|
TestClient(app) as client,
|
|
client.websocket_connect("/api/ws", headers={"X-API-Token": ""}) as websocket,
|
|
):
|
|
message = websocket.receive_json()
|
|
assert message["type"] == "status"
|
|
assert message["data"]["running"] is False
|
|
|
|
|
|
def test_api_works_without_token(runtime: Runtime) -> None:
|
|
runtime.settings.api_token = ""
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
assert client.get("/api/status").status_code == 200
|
|
assert client.get("/api/strategies").status_code == 200
|
|
assert client.post("/api/backtest", json={"strategy_key": "sma_cross"}).status_code != 401
|
|
|
|
|
|
def test_websocket_rejects_missing_token(runtime: Runtime) -> None:
|
|
runtime.settings.api_token = "secret"
|
|
app = create_app(runtime.settings, runtime)
|
|
with (
|
|
TestClient(app) as client,
|
|
pytest.raises(WebSocketDisconnect) as excinfo,
|
|
client.websocket_connect("/api/ws"),
|
|
):
|
|
pass
|
|
assert excinfo.value.code == 1008
|
|
|
|
|
|
def test_websocket_accepts_header_token(runtime: Runtime) -> None:
|
|
runtime.settings.api_token = "secret"
|
|
app = create_app(runtime.settings, runtime)
|
|
with (
|
|
TestClient(app) as client,
|
|
client.websocket_connect("/api/ws", headers={"X-API-Token": "secret"}) as websocket,
|
|
):
|
|
message = websocket.receive_json()
|
|
assert message["type"] == "status"
|
|
|
|
|
|
def test_websocket_accepts_query_token(runtime: Runtime) -> None:
|
|
runtime.settings.api_token = "secret"
|
|
app = create_app(runtime.settings, runtime)
|
|
with (
|
|
TestClient(app) as client,
|
|
client.websocket_connect("/api/ws?token=secret") as websocket,
|
|
):
|
|
message = websocket.receive_json()
|
|
assert message["type"] == "status"
|
|
|
|
|
|
def test_websocket_sends_initial_status(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client, client.websocket_connect("/api/ws") as websocket:
|
|
message = websocket.receive_json()
|
|
assert message["type"] == "status"
|
|
assert message["data"]["running"] is False
|