diff --git a/README.md b/README.md index ac1c2aa..780423d 100644 --- a/README.md +++ b/README.md @@ -64,9 +64,11 @@ podman-compose down # stoppen Port 8080 wird auf allen Interfaces veröffentlicht — das Dashboard ist damit standardmäßig von anderen Rechnern erreichbar. -> **Sicherheit.** Alle Endpunkte sind lesend (`GET`, keine Steuerbefehle), aber **nicht -> authentifiziert**. Wer den Port erreicht, sieht Kontostand, offene Positionen und die -> gesamte Trade-Historie. In offenen Netzen deshalb entweder einen Reverse Proxy mit +> **Sicherheit.** Die lesenden Endpunkte sind **nicht authentifiziert** — wer den Port +> erreicht, sieht Kontostand, offene Positionen und die gesamte Trade-Historie. Dazu kommen +> die Steuerbefehle unter `/control/…` (Training anstoßen, Lernen umschalten), die sich mit +> `server.control_token` schützen lassen und das auch sollten. Gehandelt oder konfiguriert +> werden kann über HTTP in keinem Fall. In offenen Netzen deshalb einen Reverse Proxy mit > Authentifizierung davorsetzen, den Zugriff per Firewall auf bekannte Quell-IPs begrenzen > oder auf rein lokalen Zugriff zurückstellen: > @@ -220,6 +222,9 @@ Modell ist untrainiert – lerne aus bis zu 3000 historischen Kerzen vor … Vorlernen abgeschlossen: 662 neue Beobachtungen (gesamt 662), Modell einsatzbereit ``` +Nachtrainieren lässt sich jederzeit — per Knopf im Dashboard oder über +`trademind backtest --save-model`, siehe [Training aus dem Dashboard anstoßen](#training-aus-dem-dashboard-anstoßen). + Solange das Modell nicht warm ist (`warmup_samples`), entscheidet allein das Regelwerk. Die Qualität lässt sich im Status unter `strategy.learner.online_accuracy` verfolgen — das ist eine *prequentielle* Messung: erst vorhersagen, dann lernen, also keine Selbstbewertung auf @@ -326,7 +331,7 @@ Timeframes und Parameter gehören ausprobiert, bevor auch nur ein Paper-Euro fli | Endpunkt | Inhalt | |--------------|-------------------------------------------------------------| -| `/` | Dashboard: Equity, Positionen, Trades, Modellzustand | +| `/` | Dashboard: Equity, Positionen, Trades, Modellzustand, Training | | `/health` | Liveness — nutzt der Container-Healthcheck | | `/ready` | Readiness (503, solange der Bot nicht sauber läuft) | | `/status` | vollständiger Zustand als JSON | @@ -344,6 +349,51 @@ Slack als auch Discord. --- +## Training aus dem Dashboard anstoßen + +Im Abschnitt **Training** des Dashboards stehen zwei Bedienelemente: + +**Historisch nachtrainieren.** Anzahl Kerzen wählen (500 – 50 000 je Symbol), Knopf drücken. +Der Bot lädt die Historie, läuft sie mit derselben Logik wie ein Backtest durch — **ohne zu +handeln** — und speichert das Modell danach. Der Fortschritt erscheint direkt darunter: + +``` +Fertig in 4.3s: +1099 Beobachtungen (gesamt 1753), Modell einsatzbereit, gespeichert +``` + +Ein zweiter Start wird abgelehnt, solange einer läuft. Handelsdurchlauf und Nachtraining +schließen sich gegenseitig aus, der Live-Betrieb pausiert also für die paar Sekunden. +Die Labels des laufenden Betriebs bleiben davon unberührt. + +**Kontinuierliches Lernen.** Schaltet das Online-Lernen im laufenden Betrieb an und aus. +Ausgeschaltet handelt der Bot weiter, verändert das Modell aber nicht mehr — praktisch, um +einen erreichten Stand einzufrieren, ohne den Bot anzuhalten. + +Dieselben Aktionen per HTTP: + +```bash +curl -X POST localhost:8080/control/train/history -H 'Content-Type: application/json' -H "X-TradeMind-Token: $TOKEN" -d '{"bars": 5000}' +``` + +| Endpunkt | Methode | Wirkung | +|--------------------------------|---------|--------------------------------------------| +| `/control/training` | GET | Zustand des letzten/laufenden Trainings | +| `/control/train/history` | POST | `{"bars": 5000}` — Nachtraining anstoßen | +| `/control/train/live` | POST | `{"enabled": false}` — Lernen ein/aus | + +> **Absicherung.** Anders als die lesenden Endpunkte verändern diese den Zustand des Bots. +> Setze deshalb `server.control_token` (bzw. `TRADEMIND_CONTROL_TOKEN`), sobald der Port +> nicht nur lokal erreichbar ist — die Steuerbefehle verlangen ihn dann im Header +> `X-TradeMind-Token`, während `/status` und Co. offen bleiben. Ohne Token warnt der Bot +> beim Start. Komplett abschalten lässt sich die Steuerung mit `server.enable_control: false`; +> die Routen sind dann nicht vorhanden (404). +> +> ```bash +> openssl rand -hex 24 +> ``` + +--- + ## Aufbau ``` diff --git a/config/config.example.yaml b/config/config.example.yaml index 5bac676..2928eac 100644 --- a/config/config.example.yaml +++ b/config/config.example.yaml @@ -120,6 +120,13 @@ server: port: 8080 enable_metrics: true # /metrics im Prometheus-Textformat + # Steuerbefehle im Dashboard: historisches Nachtraining anstoßen und das + # kontinuierliche Lernen ein-/ausschalten. Auf false setzen für rein lesenden Betrieb. + enable_control: true + # Schützt ALLE Steuerbefehle (Header X-TradeMind-Token). Lesende Endpunkte bleiben offen. + # Dringend empfohlen, sobald der Port nicht nur lokal erreichbar ist. + control_token: ${TRADEMIND_CONTROL_TOKEN} + # ──────────────────────────────── Benachrichtigungen ──────────────────────── notifications: # Slack- oder Discord-Webhook (leer lassen = aus) diff --git a/config/trademind.env.example b/config/trademind.env.example index 00e5215..d70b06b 100644 --- a/config/trademind.env.example +++ b/config/trademind.env.example @@ -27,6 +27,11 @@ TRADEMIND_LIVE_CONFIRMATION= # Optionaler Slack-/Discord-Webhook für Trade-Meldungen TRADEMIND_WEBHOOK_URL= +# Schützt die Steuerbefehle des Dashboards (Training anstoßen, Lernen umschalten). +# Leer = ungeschützt. Sobald der Port nicht nur lokal erreichbar ist, unbedingt setzen: +# openssl rand -hex 24 +TRADEMIND_CONTROL_TOKEN= + TRADEMIND_LOG_LEVEL=INFO # Punktuelle Overrides ohne Änderung der YAML-Datei (Muster: TRADEMIND____) diff --git a/pyproject.toml b/pyproject.toml index 453d505..6f2324a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,6 +22,7 @@ dependencies = [ dev = [ "pytest>=8.0", "pytest-asyncio>=0.23", + "pytest-aiohttp>=1.0", "ruff>=0.4", ] diff --git a/src/trademind/app.py b/src/trademind/app.py index ccf4f50..11d43d9 100644 --- a/src/trademind/app.py +++ b/src/trademind/app.py @@ -80,6 +80,24 @@ class Runtime: self.storage.close() +def _warn_about_exposure(config: Config) -> None: + """Warnen, wenn Steuerbefehle ungeschützt aus dem Netz erreichbar sind.""" + if not config.server.enable_control: + return + if config.server.control_token: + log.info("Steuerbefehle sind aktiv und durch ein Token geschützt") + return + if config.server.publicly_reachable: + log.warning( + "Steuerbefehle (Training anstoßen, Lernen umschalten) sind ohne Token auf %s:%d " + "erreichbar. Jeder, der den Port erreicht, kann sie auslösen – bitte " + "server.control_token setzen oder den Zugriff per Firewall/Reverse Proxy begrenzen.", + config.server.host, config.server.port, + ) + else: + log.info("Steuerbefehle sind aktiv (nur lokal erreichbar, kein Token gesetzt)") + + def _quote_currency(market_info: dict[str, dict[str, Any]], symbols: list[str], fallback: str) -> str: quotes = {market_info.get(s, {}).get("quote") for s in symbols} quotes.discard(None) @@ -147,7 +165,10 @@ async def build_runtime( notifier=notifier, ) - server = StatusServer(config.server, engine.status) if (with_server and config.server.enabled) else None + server = None + if with_server and config.server.enabled: + server = StatusServer(config.server, engine.status, controller=engine) + _warn_about_exposure(config) storage.start_run( mode=config.mode.value, @@ -203,4 +224,9 @@ def describe_config(config: Config) -> str: ) if config.server.enabled: lines.append(f"Status-Server http://{config.server.host}:{config.server.port}/") + if config.server.enable_control: + guard = "Token gesetzt" if config.server.control_token else "OHNE Token" + lines.append(f"Steuerung Training über das Dashboard möglich ({guard})") + else: + lines.append("Steuerung deaktiviert (nur lesende Endpunkte)") return "\n".join(" " + line for line in lines) diff --git a/src/trademind/config.py b/src/trademind/config.py index cf2fdec..1ebeed9 100644 --- a/src/trademind/config.py +++ b/src/trademind/config.py @@ -156,6 +156,15 @@ class ServerConfig(_Base): host: str = "0.0.0.0" port: int = Field(default=8080, ge=1, le=65535) enable_metrics: bool = True + # Steuerbefehle (Training anstoßen, Lernen ein-/ausschalten) über das Dashboard. + enable_control: bool = True + # Wenn gesetzt, verlangen alle Steuerbefehle den Header X-TradeMind-Token. + # Dringend empfohlen, sobald der Port nicht nur lokal erreichbar ist. + control_token: str | None = None + + @property + def publicly_reachable(self) -> bool: + return self.host not in ("127.0.0.1", "localhost", "::1") class NotificationConfig(_Base): diff --git a/src/trademind/engine.py b/src/trademind/engine.py index b845d13..513a8c2 100644 --- a/src/trademind/engine.py +++ b/src/trademind/engine.py @@ -7,10 +7,11 @@ Backtest-Betrieb; nur die Datenquelle und die Ausführung werden ausgetauscht. from __future__ import annotations import asyncio +import contextlib import logging import signal import time -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Any import numpy as np @@ -30,6 +31,85 @@ log = logging.getLogger(__name__) STATE_KEY = "engine_state" +# Grenzen für ein über das Dashboard angestoßenes Training. +MIN_TRAINING_BARS = 500 +MAX_TRAINING_BARS = 50_000 + + +@dataclass +class TrainingJob: + """Zustand eines historischen Nachtrainings – wird im Dashboard angezeigt.""" + + state: str = "idle" # idle | queued | running | done | error + bars_requested: int = 0 + bars_seen: int = 0 + symbols: list[str] = field(default_factory=list) + symbols_done: list[str] = field(default_factory=list) + skipped: list[str] = field(default_factory=list) + samples_before: int = 0 + samples_after: int = 0 + model_ready: bool = False + model_saved: bool = False + error: str | None = None + started_at: float | None = None + finished_at: float | None = None + + @property + def samples_gained(self) -> int: + return max(0, self.samples_after - self.samples_before) + + def queue(self, bars: int, symbols: list[str]) -> None: + self.state = "queued" + self.bars_requested = bars + self.bars_seen = 0 + self.symbols = list(symbols) + self.symbols_done = [] + self.skipped = [] + self.samples_before = 0 + self.samples_after = 0 + self.model_saved = False + self.error = None + self.started_at = time.time() + self.finished_at = None + + def start(self, bars: int, symbols: list[str]) -> None: + if self.state != "queued": + self.queue(bars, symbols) + self.state = "running" + + def finish(self, before: int, after: int, ready: bool) -> None: + self.samples_before = before + self.samples_after = after + self.model_ready = ready + self.state = "done" + self.finished_at = time.time() + + def fail(self, message: str) -> None: + self.state = "error" + self.error = message + self.finished_at = time.time() + + def as_dict(self) -> dict[str, Any]: + duration = ( + round((self.finished_at or time.time()) - self.started_at, 1) + if self.started_at is not None + else None + ) + return { + "state": self.state, + "bars_requested": self.bars_requested, + "bars_seen": self.bars_seen, + "symbols": self.symbols, + "symbols_done": self.symbols_done, + "skipped": self.skipped, + "samples_gained": self.samples_gained, + "samples_total": self.samples_after, + "model_ready": self.model_ready, + "model_saved": self.model_saved, + "error": self.error, + "duration_seconds": duration, + } + @dataclass(slots=True) class Bar: @@ -87,6 +167,10 @@ class TradingEngine: self._stop_event: asyncio.Event | None = None self._persist_every = 10 self._since_persist = 0 + # Handelsdurchlauf und historisches Nachtraining schließen sich gegenseitig aus. + self._engine_lock = asyncio.Lock() + self._training_task: asyncio.Task[TrainingJob] | None = None + self.training = TrainingJob() # ------------------------------------------------------------ Lebenszyklus @@ -104,6 +188,11 @@ class TradingEngine: ) async def shutdown(self, liquidate: bool = False) -> None: + if self._training_task is not None and not self._training_task.done(): + log.info("Laufendes Nachtraining wird abgebrochen …") + self._training_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self._training_task if liquidate and self.portfolio.positions: log.info("Schließe %d offene Position(en) …", len(self.portfolio.positions)) for symbol in list(self.portfolio.positions): @@ -120,45 +209,133 @@ class TradingEngine: self.running = False async def bootstrap_learner(self) -> None: - """Ein noch untrainiertes Modell aus der Kurshistorie vorlernen. + """Beim Start ein noch untrainiertes Modell aus der Kurshistorie vorlernen. Ohne diesen Schritt bräuchte ein frisch gestarteter Bot bei 5-Minuten-Kerzen mehrere Tage, bis das Modell genug Beobachtungen für die Aufwärmphase gesammelt hat. """ - learner = getattr(self.strategy, "learner", None) - warmup = getattr(self.strategy, "warmup_from_history", None) + learner = self.learner bars = self.config.strategy.learner.bootstrap_bars - if learner is None or warmup is None or not bars or learner.ready: + if learner is None or not bars or learner.ready: return - log.info("Modell ist untrainiert – lerne aus bis zu %d historischen Kerzen vor …", bars) - before = learner.stats.samples_seen - for symbol in self.config.market.symbols: - try: - candles = await self._fetch_history(symbol, bars) - except Exception as exc: # noqa: BLE001 - Vorlernen darf den Start nie verhindern - log.warning("%s: Historie für das Vorlernen nicht abrufbar (%s)", symbol, exc) - continue - matrix = build_feature_matrix(candles, self.config.strategy.rules) - if matrix is None: - log.warning("%s: zu wenig Historie zum Vorlernen (%d Kerzen)", symbol, len(candles)) - continue - last_index = warmup(symbol, matrix, candles) - # Zähler und Zeitstempel fortschreiben, damit der Live-Loop nahtlos anschließt - # und die zuletzt genutzte Kerze nicht doppelt verarbeitet wird. - self.bar_counter[symbol] = last_index - self.last_bar_ts[symbol] = int(candles.timestamp[last_index]) + await self._train_on_history(bars, adopt_timeline=True) - gained = learner.stats.samples_seen - before - log.info( - "Vorlernen abgeschlossen: %d neue Beobachtungen (gesamt %d), Modell %s", - gained, learner.stats.samples_seen, "einsatzbereit" if learner.ready else "noch im Aufwärmen", - ) - if gained: + async def _train_on_history(self, bars: int, *, adopt_timeline: bool = False) -> TrainingJob: + """Das Modell auf historischen Kerzen nachtrainieren – ohne zu handeln. + + Läuft unter demselben Mutex wie der Handelsdurchlauf, damit sich beide nicht in + die Quere kommen. Die rechenintensive Schleife wandert in einen Worker-Thread, + sonst würde der Status-Server für Sekunden blockieren. + """ + learner = self.learner + warmup = getattr(self.strategy, "warmup_from_history", None) + job = self.training + if learner is None or warmup is None: + job.fail("Kein lernfähiges Modell konfiguriert (strategy.name/learner.enabled prüfen)") + return job + + async with self._engine_lock: + job.start(bars, self.config.market.symbols) + before = learner.stats.samples_seen try: - learner.save() - except OSError as exc: # pragma: no cover - log.error("Vorgelerntes Modell konnte nicht gespeichert werden: %s", exc) + for symbol in self.config.market.symbols: + try: + candles = await self._fetch_history(symbol, bars) + except Exception as exc: # noqa: BLE001 - darf den Bot nie stoppen + log.warning("%s: Historie nicht abrufbar (%s)", symbol, exc) + job.skipped.append(f"{symbol}: {exc}") + continue + matrix = build_feature_matrix(candles, self.config.strategy.rules) + if matrix is None: + log.warning("%s: zu wenig Historie (%d Kerzen)", symbol, len(candles)) + job.skipped.append(f"{symbol}: nur {len(candles)} Kerzen") + continue + + last_index = await asyncio.to_thread(warmup, symbol, matrix, candles) + job.symbols_done.append(symbol) + job.bars_seen += len(matrix) - matrix.first_valid + if adopt_timeline: + # Zähler fortschreiben, damit der Live-Loop nahtlos anschließt und + # die zuletzt genutzte Kerze nicht doppelt verarbeitet wird. + self.bar_counter[symbol] = last_index + self.last_bar_ts[symbol] = int(candles.timestamp[last_index]) + + job.finish(before, learner.stats.samples_seen, learner.ready) + log.info( + "Historisches Training abgeschlossen: %d neue Beobachtungen (gesamt %d), Modell %s", + job.samples_gained, learner.stats.samples_seen, + "einsatzbereit" if learner.ready else "noch im Aufwärmen", + ) + if job.samples_gained: + try: + learner.save() + job.model_saved = True + except OSError as exc: # pragma: no cover - Dateisystemfehler + log.error("Modell konnte nicht gespeichert werden: %s", exc) + job.skipped.append(f"Speichern fehlgeschlagen: {exc}") + except asyncio.CancelledError: + job.fail("abgebrochen") + raise + except Exception as exc: # noqa: BLE001 - Fehler gehören in den Job-Status + log.exception("Historisches Training fehlgeschlagen") + job.fail(f"{type(exc).__name__}: {exc}") + return job + + # ----------------------------------------------------- Steuerung (Dashboard) + + @property + def learner(self): # noqa: ANN201 - AdaptiveLearner | NullLearner | None + return getattr(self.strategy, "learner", None) + + def start_history_training(self, bars: int | None = None) -> dict[str, Any]: + """Historisches Training im Hintergrund anstoßen (Aufruf kehrt sofort zurück).""" + if not self.config.server.enable_control: + return {"accepted": False, "reason": "Steuerung ist deaktiviert", **self.training.as_dict()} + # "queued" zählt mit: zwischen Anstoßen und Start des Tasks darf kein zweiter durch. + if self.training.state in ("queued", "running"): + return {"accepted": False, "reason": "Ein Training läuft bereits", **self.training.as_dict()} + if self.learner is None: + return {"accepted": False, "reason": "Kein lernfähiges Modell konfiguriert", + **self.training.as_dict()} + + requested = int(bars or self.config.strategy.learner.bootstrap_bars or 3_000) + requested = max(MIN_TRAINING_BARS, min(requested, MAX_TRAINING_BARS)) + self.training.queue(requested, self.config.market.symbols) + self._training_task = asyncio.create_task(self._train_on_history(requested)) + self._training_task.add_done_callback(lambda _: setattr(self, "_training_task", None)) + log.info("Historisches Training angefordert: %d Kerzen je Symbol", requested) + return {"accepted": True, "reason": "", **self.training.as_dict()} + + def set_online_learning(self, enabled: bool) -> dict[str, Any]: + """Kontinuierliches Lernen im laufenden Betrieb ein- oder ausschalten.""" + if not self.config.server.enable_control: + return {"accepted": False, "reason": "Steuerung ist deaktiviert", + "online_learning": self.online_learning_enabled} + learner = self.learner + if learner is None: + return {"accepted": False, "reason": "Kein lernfähiges Modell konfiguriert", + "online_learning": False} + learner.frozen = not enabled + log.info("Kontinuierliches Lernen %s", "eingeschaltet" if enabled else "eingefroren") + return {"accepted": True, "reason": "", "online_learning": enabled} + + @property + def online_learning_enabled(self) -> bool: + learner = self.learner + return bool(learner is not None and not learner.frozen) + + def training_status(self) -> dict[str, Any]: + learner = self.learner + return { + "control_enabled": self.config.server.enable_control, + "learning_available": learner is not None, + "online_learning": self.online_learning_enabled, + "default_bars": self.config.strategy.learner.bootstrap_bars or 3_000, + "min_bars": MIN_TRAINING_BARS, + "max_bars": MAX_TRAINING_BARS, + **self.training.as_dict(), + } async def _fetch_history(self, symbol: str, bars: int) -> Candles: """Längere Historie holen, wenn der Feed das kann – sonst das normale Fenster.""" @@ -208,6 +385,11 @@ class TradingEngine: log.info("Handels-Loop beendet nach %d Durchläufen (%d Fehler)", self.iterations, self.errors) async def _tick(self, history: int) -> None: + # Wartet, falls gerade ein historisches Nachtraining läuft. + async with self._engine_lock: + await self._tick_locked(history) + + async def _tick_locked(self, history: int) -> None: self._cash = await self.broker.cash() for symbol in self.config.market.symbols: candles = await self.feed.fetch(symbol, self.config.market.timeframe, history) @@ -456,6 +638,7 @@ class TradingEngine: "recent_trades": self.portfolio.recent_trades(25), "strategy": self.strategy.snapshot(), "risk": self.risk.snapshot(), + "training": self.training_status(), "feature_weights": ( learner.feature_importance(FEATURE_NAMES) if learner is not None else {} ), diff --git a/src/trademind/learner.py b/src/trademind/learner.py index e843edd..7193e6f 100644 --- a/src/trademind/learner.py +++ b/src/trademind/learner.py @@ -235,7 +235,11 @@ class LearnerStats: @dataclass(slots=True) class PendingLabel: - """Ein Kandidatensignal, dessen Ausgang erst in der Zukunft feststeht.""" + """Ein Kandidatensignal, dessen Ausgang erst in der Zukunft feststeht. + + ``tag`` trennt Läufe voneinander: Ein historisches Nachtraining darf die Labels des + laufenden Live-Betriebs weder auflösen noch verwerfen. + """ symbol: str features: np.ndarray @@ -244,6 +248,7 @@ class PendingLabel: horizon_bars: int target_bps: float weight: float = 1.0 + tag: str = "live" def matured(self, current_bar: int) -> bool: return current_bar - self.created_bar >= self.horizon_bars @@ -275,9 +280,14 @@ class AdaptiveLearner: return self.stats.samples_seen >= self.config.warmup_samples def score(self, features: np.ndarray) -> float: - """Gewinnwahrscheinlichkeit für ein Einstiegssignal (0..1).""" - x = self.scaler.transform(np.asarray(features, dtype=np.float64).reshape(1, -1)) - return float(self.model.predict_proba(x)[0]) + """Gewinnwahrscheinlichkeit für ein Einstiegssignal (0..1). + + Unter dem Lock, damit ein parallel laufendes Nachtraining keinen halb + aktualisierten Gewichtsvektor sichtbar macht. + """ + with self._lock: + x = self.scaler.transform(np.asarray(features, dtype=np.float64).reshape(1, -1)) + return float(self.model.predict_proba(x)[0]) def explore(self) -> bool: """Epsilon-greedy: gelegentlich ein abgelehntes Signal trotzdem handeln.""" @@ -345,7 +355,13 @@ class AdaptiveLearner: # -------------------------------------------------- Verzögerte Shadow-Labels def register_candidate( - self, symbol: str, features: np.ndarray, price: float, bar_index: int, weight: float = 1.0 + self, + symbol: str, + features: np.ndarray, + price: float, + bar_index: int, + weight: float = 1.0, + tag: str = "live", ) -> None: """Kandidatensignal vormerken; das Label folgt nach ``label_horizon_bars``.""" self._pending.append( @@ -357,21 +373,26 @@ class AdaptiveLearner: horizon_bars=self.config.label_horizon_bars, target_bps=self.config.label_target_bps, weight=weight, + tag=tag, ) ) - def resolve_pending(self, symbol: str, bar_index: int, high: float, low: float, close: float) -> int: + def resolve_pending( + self, symbol: str, bar_index: int, high: float, low: float, close: float, tag: str | None = None + ) -> int: """Fällige Shadow-Labels auswerten. Gibt die Anzahl neuer Beobachtungen zurück. Label = 1, wenn der Kurs innerhalb des Horizonts das Ziel erreicht hat, ohne vorher um denselben Betrag zu fallen (vereinfachte Triple-Barrier-Methode). + + ``tag`` grenzt auf einen Lauf ein; ``None`` wertet alle aus. """ if not self._pending: return 0 resolved = 0 still_open: list[PendingLabel] = [] for item in self._pending: - if item.symbol != symbol: + if item.symbol != symbol or (tag is not None and item.tag != tag): still_open.append(item) continue target = item.entry_price * (1.0 + item.target_bps / 10_000.0) @@ -393,8 +414,15 @@ class AdaptiveLearner: self._pending = still_open return resolved - def drop_pending(self, symbol: str | None = None) -> None: - self._pending = [p for p in self._pending if symbol is not None and p.symbol != symbol] + def drop_pending(self, symbol: str | None = None, tag: str | None = None) -> int: + """Vorgemerkte Labels verwerfen. Ohne Argumente alle, sonst nur die passenden.""" + + def matches(item: PendingLabel) -> bool: + return (symbol is None or item.symbol == symbol) and (tag is None or item.tag == tag) + + before = len(self._pending) + self._pending = [p for p in self._pending if not matches(p)] + return before - len(self._pending) @property def pending_count(self) -> int: @@ -537,8 +565,8 @@ class NullLearner: def resolve_pending(self, *args: object, **kwargs: object) -> int: return 0 - def drop_pending(self, *args: object, **kwargs: object) -> None: - return None + def drop_pending(self, *args: object, **kwargs: object) -> int: + return 0 def learn_from_trade(self, *args: object, **kwargs: object) -> None: return None diff --git a/src/trademind/server.py b/src/trademind/server.py index ef9d5b4..8d1c61f 100644 --- a/src/trademind/server.py +++ b/src/trademind/server.py @@ -2,9 +2,10 @@ from __future__ import annotations +import hmac import logging from collections.abc import Callable -from typing import Any +from typing import Any, Protocol from aiohttp import web @@ -14,6 +15,18 @@ log = logging.getLogger(__name__) StatusProvider = Callable[[], dict[str, Any]] +TOKEN_HEADER = "X-TradeMind-Token" + + +class Controller(Protocol): + """Was der Server zum Steuern braucht – von :class:`~trademind.engine.TradingEngine` erfüllt.""" + + def start_history_training(self, bars: int | None = None) -> dict[str, Any]: ... + + def set_online_learning(self, enabled: bool) -> dict[str, Any]: ... + + def training_status(self) -> dict[str, Any]: ... + _DASHBOARD = """ TradeMind @@ -35,10 +48,49 @@ _DASHBOARD = """ th:first-child, td:first-child { text-align: left; } section { margin-top: 28px; } .overflow { overflow-x: auto; } + h2 { font-size: 15px; margin: 0 0 6px; } + .panel { background: var(--card); border: 1px solid var(--line); border-radius: 10px; padding: 14px 16px; } + .row { display: flex; flex-wrap: wrap; gap: 10px; align-items: center; } + button { font: inherit; padding: 7px 14px; border-radius: 7px; border: 1px solid var(--line); + background: var(--fg); color: var(--bg); cursor: pointer; } + button.ghost { background: transparent; color: var(--fg); } + button:disabled { opacity: .5; cursor: not-allowed; } + input { font: inherit; padding: 6px 9px; border-radius: 7px; border: 1px solid var(--line); + background: var(--bg); color: var(--fg); } + input[type=number] { width: 8em; } + label { color: var(--muted); font-size: 13px; } + .hint { color: var(--muted); font-size: 12px; margin-top: 8px; } + .hint.err { color: #c62828; } + .dot { display: inline-block; width: 8px; height: 8px; border-radius: 50%; margin-right: 6px; } + .dot.on { background: #1a8f3c; } .dot.off { background: #999; } .dot.busy { background: #e0a100; }

TradeMind

lädt …
+ +

Offene Positionen

Letzte Trades

""" @@ -101,11 +233,21 @@ def _flatten_metrics(prefix: str, node: Any, out: list[tuple[str, float]]) -> No class StatusServer: """Kleiner aiohttp-Server; hält keinen eigenen Zustand, sondern fragt den Bot ab.""" - def __init__(self, config: ServerConfig, status_provider: StatusProvider) -> None: + def __init__( + self, + config: ServerConfig, + status_provider: StatusProvider, + controller: Controller | None = None, + ) -> None: self.config = config self._status = status_provider + self._controller = controller self._runner: web.AppRunner | None = None + @property + def control_available(self) -> bool: + return self.config.enable_control and self._controller is not None + def _build_app(self) -> web.Application: app = web.Application() app.add_routes( @@ -120,6 +262,14 @@ class StatusServer: ) if self.config.enable_metrics: app.router.add_get("/metrics", self._metrics) + if self.control_available: + app.add_routes( + [ + web.get("/control/training", self._training_state), + web.post("/control/train/history", self._train_history), + web.post("/control/train/live", self._train_live), + ] + ) return app async def start(self) -> None: @@ -162,6 +312,70 @@ class StatusServer: trades = self._status().get("recent_trades", []) return web.json_response(trades[-limit:], dumps=_dumps) + # ------------------------------------------------------------- Steuerung + + def _authorized(self, request: web.Request) -> bool: + token = self.config.control_token + if not token: + return True + supplied = request.headers.get(TOKEN_HEADER, "") + return hmac.compare_digest(supplied, token) + + async def _guard(self, request: web.Request) -> web.Response | None: + """Gibt eine Fehlerantwort zurück, wenn der Aufruf nicht erlaubt ist.""" + if self._controller is None or not self.config.enable_control: + return web.json_response({"error": "Steuerung ist deaktiviert"}, status=404) + if not self._authorized(request): + log.warning("Steuerbefehl ohne gültiges Token abgelehnt (%s)", request.remote) + return web.json_response( + {"error": f"Ungültiges oder fehlendes Token im Header {TOKEN_HEADER}"}, status=401 + ) + return None + + async def _training_state(self, request: web.Request) -> web.Response: + denied = await self._guard(request) + if denied is not None: + return denied + assert self._controller is not None + return web.json_response(self._controller.training_status(), dumps=_dumps) + + async def _train_history(self, request: web.Request) -> web.Response: + denied = await self._guard(request) + if denied is not None: + return denied + assert self._controller is not None + + bars: int | None = None + if request.can_read_body: + try: + payload = await request.json() + except ValueError: + return web.json_response({"error": "Ungültiges JSON"}, status=400) + if isinstance(payload, dict) and payload.get("bars") is not None: + try: + bars = int(payload["bars"]) + except (TypeError, ValueError): + return web.json_response({"error": "'bars' muss eine Zahl sein"}, status=400) + + result = self._controller.start_history_training(bars) + return web.json_response(result, status=202 if result.get("accepted") else 409, dumps=_dumps) + + async def _train_live(self, request: web.Request) -> web.Response: + denied = await self._guard(request) + if denied is not None: + return denied + assert self._controller is not None + + try: + payload = await request.json() + except ValueError: + return web.json_response({"error": "Ungültiges JSON"}, status=400) + if not isinstance(payload, dict) or not isinstance(payload.get("enabled"), bool): + return web.json_response({"error": "'enabled' (true/false) wird erwartet"}, status=400) + + result = self._controller.set_online_learning(payload["enabled"]) + return web.json_response(result, status=200 if result.get("accepted") else 409, dumps=_dumps) + async def _metrics(self, _: web.Request) -> web.Response: state = self._status() samples: list[tuple[str, float]] = [] diff --git a/src/trademind/strategy.py b/src/trademind/strategy.py index 06fedba..7261199 100644 --- a/src/trademind/strategy.py +++ b/src/trademind/strategy.py @@ -115,6 +115,7 @@ class AdaptiveStrategy(Strategy): self.candidates_explored = 0 self.background_samples = 0 self._bar_index: dict[str, int] = {} + self._label_tag = "live" def evaluate(self, symbol: str, snapshot: FeatureSnapshot, position: Position | None) -> Signal: if position is not None: @@ -163,12 +164,14 @@ class AdaptiveStrategy(Strategy): self, symbol: str, snapshot: FeatureSnapshot, bar_index: int, weight: float = 1.0 ) -> None: """Kandidaten für das verzögerte Labeln vormerken (auch abgelehnte).""" - self.learner.register_candidate(symbol, snapshot.values, snapshot.price, bar_index, weight) + self.learner.register_candidate( + symbol, snapshot.values, snapshot.price, bar_index, weight, tag=self._label_tag + ) def on_bar(self, symbol: str, snapshot: FeatureSnapshot, bar_index: int, high: float, low: float, close: float) -> None: self._bar_index[symbol] = bar_index - self.learner.resolve_pending(symbol, bar_index, high, low, close) + self.learner.resolve_pending(symbol, bar_index, high, low, close, tag=self._label_tag) # Einstiegssignale sind selten – regelmäßige Stichproben des Marktzustands geben # dem Modell genug Daten, um die Aufwärmphase in vertretbarer Zeit zu durchlaufen. @@ -186,26 +189,43 @@ class AdaptiveStrategy(Strategy): def warmup_from_history(self, symbol: str, matrix: FeatureMatrix, candles: Candles) -> int: """Trainiert das Modell offline auf vorhandener Kurshistorie. - Damit ist ein frisch ausgerollter Bot nach Sekunden einsatzbereit statt nach Tagen. - Es werden ausschließlich vergangene Kerzen verwendet – dieselbe Logik wie im Backtest. - Gibt den Index der letzten verarbeiteten Kerze zurück. + Damit ist ein frisch ausgerollter Bot nach Sekunden einsatzbereit statt nach Tagen, + und ein laufender Bot lässt sich jederzeit nachtrainieren. Es werden ausschließlich + vergangene Kerzen verwendet – dieselbe Logik wie im Backtest, nur ohne Handel. + + Die dabei erzeugten Labels laufen unter dem Tag ``history`` und bleiben damit von + den offenen Labels des Live-Betriebs getrennt. Gibt den Index der letzten + verarbeiteten Kerze zurück. """ + previous_tag = self._label_tag + previous_bar_index = self._bar_index.get(symbol) + self._label_tag = "history" last_index = matrix.first_valid - for index in range(matrix.first_valid, len(matrix)): - snapshot = matrix.snapshot(index) - if snapshot is None: - continue - self.on_bar( - symbol, - snapshot, - index, - float(candles.high[index]), - float(candles.low[index]), - float(candles.close[index]), - ) - if self.rules.entry_candidate(snapshot) is not None: - self.register_candidate(symbol, snapshot, index) - last_index = index + try: + for index in range(matrix.first_valid, len(matrix)): + snapshot = matrix.snapshot(index) + if snapshot is None: + continue + self.on_bar( + symbol, + snapshot, + index, + float(candles.high[index]), + float(candles.low[index]), + float(candles.close[index]), + ) + if self.rules.entry_candidate(snapshot) is not None: + self.register_candidate(symbol, snapshot, index) + last_index = index + finally: + self._label_tag = previous_tag + # Nicht aufgelöste Historien-Labels verwerfen; sie würden sonst gegen die + # nächste Live-Kerze ausgewertet und das Modell mit falschen Labels füttern. + self.learner.drop_pending(symbol=symbol, tag="history") + if previous_bar_index is None: + self._bar_index.pop(symbol, None) + else: + self._bar_index[symbol] = previous_bar_index return last_index def snapshot(self) -> dict[str, object]: diff --git a/tests/test_control.py b/tests/test_control.py new file mode 100644 index 0000000..928c4a8 --- /dev/null +++ b/tests/test_control.py @@ -0,0 +1,321 @@ +"""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.config import Config, ServerConfig +from trademind.features import N_FEATURES +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 + + +# --------------------------------------------------------------- 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 + + +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