admin/Trademind
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity
Files
4551339632af432a9c8d07b8aa6a4ea094c9c16a
Trademind/tests/test_control.py
T
Tobias Zimmermann d081edfbd3 Automatisierten Handel über das Dashboard starten und pausieren 
Neuer Abschnitt "Automatisierter Handel" im Dashboard mit Zustandsanzeige und
Start/Pause-Knopf, dazu GET und POST /control/trading. Funktioniert in beiden
Modi, simuliert wie echt.

Pausiert unterbindet ausschließlich neue Einstiege
- Marktdaten, Signalauswertung, Labeling und Training laufen weiter, damit der
  Datenstrom ins Modell nie abreißt. Auf identischen Daten: aktiv 99
  Beobachtungen bei 9 Trades, pausiert 90 Beobachtungen bei 0 Trades - der
  Unterschied sind genau die Trade-Labels, die ohne Handel nicht entstehen.
- Offene Positionen bleiben unter Stop- und Ziel-Überwachung. Ein pausierter Bot
  lässt niemanden ungeschützt im Markt stehen.
- Neu: trading.autostart (Standard true, bisheriges Verhalten) und
  trading.require_confirmation_for_live.

Nebenbei behoben
- Bei aktiver Notbremse (Tagesverlust, Drawdown) wurden bisher gar keine Signale
  mehr ausgewertet. Der Bot hörte damit ausgerechnet in der interessantesten
  Phase auf zu lernen. Jetzt läuft die Auswertung durch, nur der Kauf unterbleibt.
- describe_config meldete "Steuerung möglich", während sie eine Zeile später
  abgeschaltet wurde. Übersicht und Schutzlogik teilen sich jetzt dieselbe
  Prüffunktion (control_effective / control_is_unsafe).

Absicherung für Echtgeld
- Starten im Live-Modus verlangt die Bestätigung START_LIVE_TRADING im Body
  (HTTP 428 ohne sie); das Dashboard fragt sie per Dialog ab. Pausieren geht
  immer ohne Bestätigung - anhalten muss schnell gehen.
- Live-Modus mit offenem Port und ohne control_token wird nicht bedient: Die
  Steuerung wird beim Start abgeschaltet, der Bot läuft normal weiter. Sonst
  könnte jeder im Netz echten Handel starten.

Bewusst nicht enthalten: ein Umschalter zwischen Paper und Live zur Laufzeit.
Der bräuchte einen neuen Broker, andere Zugangsdaten und eine neue Kapitalbasis -
und würde bedeuten, dass ein Web-Klick aus einer Simulation echten Handel macht.

174 Tests (21 neue), ruff sauber. Im gebauten Container geprüft: 401 ohne Token,
200 mit, idempotentes Pausieren, 400 bei kaputtem Payload, und im Live-Start mit
ungültigen Dummy-Schlüsseln greift die Abschaltung der Steuerung.
2026-08-23 12:40:56 +02:00

561 lines
21 KiB
Python
Raw Blame History

"""Steuerung über das Dashboard: historisches Nachtraining und Live-Lernschalter."""
from __future__ import annotations
import asyncio
import numpy as np
import pytest
from aiohttp import web
from trademind.app import _guard_control_exposure
from trademind.backtest import BacktestRunner
from trademind.config import LIVE_CONFIRMATION_PHRASE, Config, ServerConfig
from trademind.engine import LIVE_TRADING_CONFIRMATION, Bar
from trademind.features import N_FEATURES, compute_features
from trademind.models import ExitReason, Side
from trademind.server import TOKEN_HEADER, StatusServer
from .conftest import make_candles
from .test_engine import GrowingFeed, StaticFeed, build_engine, cyclical_series
def control_config(base_config: Config, **server: object) -> Config:
return Config.model_validate(
{**base_config.model_dump(), "mode": "paper", "server": {"enabled": True, **server}}
)
# ------------------------------------------------------- Historisches Training
async def test_history_training_can_be_started_and_completes(base_config):
engine = build_engine(control_config(base_config), feed=GrowingFeed(cyclical_series(n=900), start=900))
await engine.prepare()
result = engine.start_history_training(bars=900)
assert result["accepted"] is True
assert engine.training.state == "queued"
await engine._training_task
job = engine.training
assert job.state == "done"
assert job.samples_gained > 0
assert job.symbols_done == ["BTC/USDT"]
assert job.bars_seen > 0
assert engine.strategy.learner.stats.samples_seen == job.as_dict()["samples_total"]
async def test_bars_are_clamped_to_sane_limits(base_config):
engine = build_engine(control_config(base_config), feed=GrowingFeed(cyclical_series(n=900), start=900))
await engine.prepare()
engine.start_history_training(bars=10)
assert engine.training.bars_requested == 500 # Untergrenze
await engine._training_task
engine.start_history_training(bars=10_000_000)
assert engine.training.bars_requested == 50_000 # Obergrenze
await engine._training_task
async def test_second_training_is_rejected_while_one_runs(base_config):
engine = build_engine(control_config(base_config), feed=GrowingFeed(cyclical_series(n=900), start=900))
await engine.prepare()
engine.start_history_training(bars=900)
second = engine.start_history_training(bars=900)
assert second["accepted"] is False
assert "läuft bereits" in second["reason"]
await engine._training_task
async def test_training_does_not_trade(base_config):
engine = build_engine(control_config(base_config), feed=GrowingFeed(cyclical_series(n=900), start=900))
await engine.prepare()
engine.start_history_training(bars=900)
await engine._training_task
assert engine.portfolio.trades == []
assert engine.portfolio.positions == {}
assert await engine.broker.cash() == pytest.approx(base_config.paper.starting_balance)
async def test_training_leaves_live_labels_untouched(base_config):
"""Historisches Training darf offene Live-Labels weder auflösen noch verwerfen."""
engine = build_engine(control_config(base_config), feed=GrowingFeed(cyclical_series(n=900), start=900))
await engine.prepare()
learner = engine.strategy.learner
learner.register_candidate("BTC/USDT", np.ones(N_FEATURES), 100.0, bar_index=5, tag="live")
assert learner.pending_count == 1
engine.start_history_training(bars=900)
await engine._training_task
remaining = [p for p in learner._pending if p.tag == "live"]
assert len(remaining) == 1
assert remaining[0].entry_price == 100.0
assert not [p for p in learner._pending if p.tag == "history"]
async def test_training_does_not_shift_the_live_timeline(base_config):
engine = build_engine(control_config(base_config), feed=GrowingFeed(cyclical_series(n=900), start=900))
await engine.prepare()
engine.bar_counter["BTC/USDT"] = 42
engine.last_bar_ts["BTC/USDT"] = 1_234_567
engine.start_history_training(bars=900)
await engine._training_task
assert engine.bar_counter["BTC/USDT"] == 42
assert engine.last_bar_ts["BTC/USDT"] == 1_234_567
async def test_bootstrap_still_adopts_the_timeline(base_config):
"""Beim Kaltstart soll der Live-Loop dagegen an die Historie anschließen."""
engine = build_engine(control_config(base_config), feed=GrowingFeed(cyclical_series(n=900), start=900))
await engine.prepare()
await engine.bootstrap_learner()
assert engine.bar_counter["BTC/USDT"] > 0
async def test_training_and_tick_do_not_overlap(base_config):
engine = build_engine(control_config(base_config), feed=GrowingFeed(cyclical_series(n=900), start=900))
await engine.prepare()
engine.start_history_training(bars=900)
tick = asyncio.create_task(engine._tick(300))
await asyncio.gather(engine._training_task, tick)
assert engine.training.state == "done"
assert engine.errors == 0
async def test_training_failure_is_reported_not_raised(base_config):
class BrokenFeed(StaticFeed):
async def fetch(self, symbol, timeframe, limit):
raise RuntimeError("Börse nicht erreichbar")
engine = build_engine(control_config(base_config), feed=BrokenFeed())
await engine.prepare()
engine.start_history_training(bars=900)
await engine._training_task
job = engine.training
assert job.state == "done" # der Lauf endet geordnet
assert job.symbols_done == []
assert any("nicht erreichbar" in s for s in job.skipped)
async def test_short_history_is_skipped(base_config):
engine = build_engine(control_config(base_config), feed=StaticFeed(make_candles(n=60)))
await engine.prepare()
engine.start_history_training(bars=900)
await engine._training_task
assert any("Kerzen" in s for s in engine.training.skipped)
async def test_rules_strategy_cannot_be_trained(base_config):
config = Config.model_validate(
{**control_config(base_config).model_dump(), "strategy": {"name": "rules"}}
)
engine = build_engine(config, feed=StaticFeed())
await engine.prepare()
result = engine.start_history_training()
assert result["accepted"] is False
assert "lernfähiges Modell" in result["reason"]
# ------------------------------------------------------- Live-Lernen umschalten
async def test_online_learning_can_be_toggled(base_config):
engine = build_engine(control_config(base_config), feed=StaticFeed())
await engine.prepare()
assert engine.online_learning_enabled is True
assert engine.set_online_learning(False)["online_learning"] is False
assert engine.strategy.learner.frozen is True
assert engine.online_learning_enabled is False
assert engine.set_online_learning(True)["online_learning"] is True
assert engine.strategy.learner.frozen is False
async def test_frozen_learner_stops_updating_weights(base_config):
engine = build_engine(control_config(base_config), feed=StaticFeed())
await engine.prepare()
learner = engine.strategy.learner
for _ in range(60):
learner.observe(np.ones(N_FEATURES), 1.0)
engine.set_online_learning(False)
weights = learner.model.w.copy()
for _ in range(60):
learner.observe(np.ones(N_FEATURES), 0.0)
assert np.allclose(learner.model.w, weights)
async def test_control_disabled_rejects_commands(base_config):
engine = build_engine(control_config(base_config, enable_control=False), feed=StaticFeed())
await engine.prepare()
assert engine.start_history_training()["accepted"] is False
assert engine.set_online_learning(False)["accepted"] is False
# ------------------------------------------------------ Automatisierter Handel
def trading_config(base_config: Config, **trading: object) -> Config:
return Config.model_validate(
{**control_config(base_config).model_dump(), "trading": trading}
)
async def test_trading_is_active_by_default(base_config):
engine = build_engine(control_config(base_config), feed=StaticFeed())
await engine.prepare()
assert engine.trading_active is True
assert engine.trading_control_status()["simulated"] is True
async def test_autostart_false_starts_paused(base_config):
engine = build_engine(trading_config(base_config, autostart=False), feed=StaticFeed())
await engine.prepare()
assert engine.trading_active is False
async def test_trading_can_be_paused_and_resumed(base_config):
engine = build_engine(control_config(base_config), feed=StaticFeed())
await engine.prepare()
assert engine.set_trading(False)["active"] is False
assert engine.trading_active is False
assert engine.set_trading(True)["active"] is True
assert engine.trading_active is True
async def test_repeated_state_is_accepted_without_change(base_config):
engine = build_engine(control_config(base_config), feed=StaticFeed())
await engine.prepare()
result = engine.set_trading(True)
assert result["accepted"] is True
assert "bereits" in result["reason"]
async def test_paused_bot_opens_no_positions(base_config):
engine = build_engine(trading_config(base_config, autostart=False))
await engine.prepare()
await BacktestRunner(engine, {"BTC/USDT": cyclical_series()}, progress_every=0).run()
assert engine.portfolio.trades == []
assert engine.portfolio.positions == {}
assert engine.paused_signals > 0, "es hätte Signale geben müssen, die nur gelernt wurden"
async def test_paused_bot_keeps_learning(base_config):
"""Kern der Anforderung: Pausiert fließen weiter Daten ins Modell."""
paused = build_engine(trading_config(base_config, autostart=False))
await paused.prepare()
await BacktestRunner(paused, {"BTC/USDT": cyclical_series()}, progress_every=0).run()
assert paused.strategy.learner.stats.samples_seen > 0
assert paused.strategy.learner.stats.shadow_samples > 0
assert paused.strategy.learner.stats.updates > 0
assert paused.strategy.candidates_seen > 0
# Ohne Trades gibt es logischerweise keine Trade-Labels, nur Shadow-Labels.
assert paused.strategy.learner.stats.trade_samples == 0
async def test_active_bot_trades_on_the_same_data(base_config):
active = build_engine(control_config(base_config))
await active.prepare()
await BacktestRunner(active, {"BTC/USDT": cyclical_series()}, progress_every=0).run()
assert active.portfolio.stats.trades > 0
async def test_pausing_keeps_stop_loss_active(base_config):
"""Eine offene Position darf beim Pausieren nicht ungeschützt stehenbleiben."""
engine = build_engine(control_config(base_config))
await engine.prepare()
fill = await engine.broker.execute("BTC/USDT", Side.BUY, 0.05, 30_000.0)
engine.portfolio.open_position(
fill, stop_loss=29_500.0, take_profit=None,
features=np.ones(N_FEATURES), confidence=0.6, exploratory=False,
)
engine.set_trading(False)
snapshot = compute_features(cyclical_series(), base_config.strategy.rules)
bar = Bar(timestamp=1, open=29_600.0, high=29_700.0, low=29_000.0, close=29_400.0, volume=5.0)
await engine.process_bar("BTC/USDT", snapshot, bar)
assert engine.portfolio.positions == {}
assert engine.portfolio.trades[-1].exit_reason is ExitReason.STOP_LOSS
async def test_live_mode_requires_confirmation(base_config, monkeypatch):
engine = build_engine(control_config(base_config), feed=StaticFeed())
await engine.prepare()
monkeypatch.setattr(type(engine.config), "is_simulated", property(lambda self: False))
engine.trading_active = False
denied = engine.set_trading(True)
assert denied["accepted"] is False
assert LIVE_TRADING_CONFIRMATION in denied["reason"]
assert engine.trading_active is False
assert engine.set_trading(True, "falsch")["accepted"] is False
assert engine.trading_active is False
ok = engine.set_trading(True, LIVE_TRADING_CONFIRMATION)
assert ok["accepted"] is True
assert engine.trading_active is True
async def test_pausing_live_never_needs_confirmation(base_config, monkeypatch):
engine = build_engine(control_config(base_config), feed=StaticFeed())
await engine.prepare()
monkeypatch.setattr(type(engine.config), "is_simulated", property(lambda self: False))
assert engine.set_trading(False)["accepted"] is True
async def test_paper_mode_needs_no_confirmation(base_config):
engine = build_engine(control_config(base_config), feed=StaticFeed())
await engine.prepare()
engine.set_trading(False)
assert engine.set_trading(True)["accepted"] is True
async def test_trading_control_rejected_when_disabled(base_config):
engine = build_engine(control_config(base_config, enable_control=False), feed=StaticFeed())
await engine.prepare()
assert engine.set_trading(False)["accepted"] is False
assert engine.trading_active is True
# --------------------------------------------------------------- HTTP-Schicht
class SlowFeed(GrowingFeed):
"""Verzögert den Abruf, damit sich Anfragen zuverlässig überlappen."""
async def fetch(self, symbol: str, timeframe: str, limit: int):
await asyncio.sleep(0.3)
return await super().fetch(symbol, timeframe, limit)
@pytest.fixture
async def client(aiohttp_client, base_config):
"""Server mit angeschlossener Engine; gibt (client, engine) zurück."""
async def _make(feed=None, **server_options: object):
config = control_config(base_config, **server_options)
engine = build_engine(config, feed=feed or GrowingFeed(cyclical_series(n=900), start=900))
await engine.prepare()
srv = StatusServer(config.server, engine.status, controller=engine)
return await aiohttp_client(srv._build_app()), engine
return _make
async def test_dashboard_and_status_expose_training(client):
http, _ = await client()
page = await (await http.get("/")).text()
assert "Training starten" in page
assert "Kontinuierliches Lernen" in page
status = await (await http.get("/status")).json()
assert status["training"]["control_enabled"] is True
assert status["training"]["online_learning"] is True
async def test_post_history_training_returns_202(client):
http, engine = await client()
response = await http.post("/control/train/history", json={"bars": 900})
assert response.status == 202
assert (await response.json())["accepted"] is True
await engine._training_task
async def test_post_history_training_conflict_returns_409(client):
http, engine = await client(feed=SlowFeed(cyclical_series(n=900), start=900))
first = await http.post("/control/train/history", json={"bars": 900})
assert first.status == 202
second = await http.post("/control/train/history", json={"bars": 900})
assert second.status == 409
assert "läuft bereits" in (await second.json())["reason"]
await engine._training_task
assert engine.training.state == "done"
async def test_post_live_toggle(client):
http, engine = await client()
response = await http.post("/control/train/live", json={"enabled": False})
assert response.status == 200
assert (await response.json())["online_learning"] is False
assert engine.strategy.learner.frozen is True
async def test_live_toggle_rejects_bad_payload(client):
http, _ = await client()
assert (await http.post("/control/train/live", json={"enabled": "ja"})).status == 400
assert (await http.post("/control/train/live", data="kein json")).status == 400
async def test_token_is_required_when_configured(client):
http, engine = await client(control_token="geheim")
assert (await http.post("/control/train/live", json={"enabled": False})).status == 401
assert (await http.get("/control/training")).status == 401
assert engine.strategy.learner.frozen is False # nichts passiert
ok = await http.post(
"/control/train/live", json={"enabled": False}, headers={TOKEN_HEADER: "geheim"}
)
assert ok.status == 200
assert engine.strategy.learner.frozen is True
async def test_wrong_token_is_rejected(client):
http, _ = await client(control_token="geheim")
response = await http.post(
"/control/train/live", json={"enabled": False}, headers={TOKEN_HEADER: "falsch"}
)
assert response.status == 401
async def test_control_routes_absent_when_disabled(client):
http, _ = await client(enable_control=False)
assert (await http.post("/control/train/history", json={})).status == 404
assert (await http.get("/control/training")).status == 404
assert (await http.get("/status")).status == 200 # lesende Endpunkte bleiben
async def test_read_only_endpoints_need_no_token(client):
http, _ = await client(control_token="geheim")
for path in ("/health", "/status", "/metrics", "/positions", "/trades"):
assert (await http.get(path)).status == 200, path
async def test_training_state_endpoint(client):
http, _ = await client()
data = await (await http.get("/control/training")).json()
assert data["state"] == "idle"
assert data["learning_available"] is True
assert data["min_bars"] == 500
async def test_http_trading_toggle(client):
http, engine = await client()
response = await http.post("/control/trading", json={"enabled": False})
assert response.status == 200
assert (await response.json())["active"] is False
assert engine.trading_active is False
state = await (await http.get("/control/trading")).json()
assert state["active"] is False
assert state["mode"] == "paper"
async def test_http_trading_rejects_bad_payload(client):
http, _ = await client()
assert (await http.post("/control/trading", json={})).status == 400
assert (await http.post("/control/trading", json={"enabled": 1})).status == 400
assert (await http.post("/control/trading", json={"enabled": True, "confirm": 5})).status == 400
async def test_http_trading_needs_token(client):
http, engine = await client(control_token="geheim")
assert (await http.post("/control/trading", json={"enabled": False})).status == 401
assert engine.trading_active is True
ok = await http.post(
"/control/trading", json={"enabled": False}, headers={TOKEN_HEADER: "geheim"}
)
assert ok.status == 200
assert engine.trading_active is False
async def test_http_live_start_without_confirmation_returns_428(client, monkeypatch):
http, engine = await client()
monkeypatch.setattr(type(engine.config), "is_simulated", property(lambda self: False))
engine.trading_active = False
response = await http.post("/control/trading", json={"enabled": True})
assert response.status == 428
assert engine.trading_active is False
ok = await http.post(
"/control/trading", json={"enabled": True, "confirm": LIVE_TRADING_CONFIRMATION}
)
assert ok.status == 200
assert engine.trading_active is True
async def test_status_and_dashboard_expose_trading(client):
http, _ = await client()
status = await (await http.get("/status")).json()
assert status["trading"]["active"] is True
assert status["trading"]["simulated"] is True
page = await (await http.get("/")).text()
assert "Automatisierter Handel" in page
assert "control/trading" in page
assert "LIVE-MODUS" in page
# ------------------------------------------- Absicherung beim Zusammenbau
def test_live_without_token_on_open_port_disables_control():
config = Config.model_validate(
{
"mode": "live",
"live_confirmation": LIVE_CONFIRMATION_PHRASE,
"exchange": {"api_key": "k", "api_secret": "s"},
"server": {"enabled": True, "host": "0.0.0.0", "enable_control": True},
}
)
guarded = _guard_control_exposure(config)
assert guarded.server.enable_control is False
def test_live_with_token_keeps_control():
config = Config.model_validate(
{
"mode": "live",
"live_confirmation": LIVE_CONFIRMATION_PHRASE,
"exchange": {"api_key": "k", "api_secret": "s"},
"server": {"enabled": True, "host": "0.0.0.0", "enable_control": True,
"control_token": "geheim"},
}
)
assert _guard_control_exposure(config).server.enable_control is True
def test_live_on_loopback_keeps_control():
config = Config.model_validate(
{
"mode": "live",
"live_confirmation": LIVE_CONFIRMATION_PHRASE,
"exchange": {"api_key": "k", "api_secret": "s"},
"server": {"enabled": True, "host": "127.0.0.1", "enable_control": True},
}
)
assert _guard_control_exposure(config).server.enable_control is True
def test_paper_on_open_port_keeps_control(base_config):
config = control_config(base_config, host="0.0.0.0", enable_control=True)
assert _guard_control_exposure(config).server.enable_control is True
def test_server_app_builds_without_controller(base_config):
srv = StatusServer(ServerConfig(), lambda: {}, controller=None)
app = srv._build_app()
assert isinstance(app, web.Application)
assert srv.control_available is False
Reference in New Issue View Git Blame Copy Permalink