matitos
03dfa369e5
- RiskManager halts + flattens on max daily loss / drawdown, caps position size and trades per day; adds a manual kill switch (POST /api/bot/kill). - Telegram and webhook notifications on trades, errors and halts. - Optional API bearer token on /api, injected server-side by the frontend nginx so the browser never sees it. - Kraken and Coinbase market data (public); Coinbase live broker. Live on an unsupported exchange now errors instead of silently falling back to paper. - Strategy presets backed by the StrategyConfig table (CRUD API + UI). - Idempotent additive-column migration so existing SQLite databases upgrade in place (fixes the new strategy_configs.name column).
152 lines
5.5 KiB
Python
152 lines
5.5 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import create_app
|
|
from app.market.fake import FakeMarketData
|
|
from app.runtime import Runtime
|
|
|
|
|
|
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_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_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_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
|