Tobias Zimmermann
65ed73977e
Per Podman deploybarer Bot, der Käufe und Verkäufe simuliert ausführt und sich aus den Ergebnissen weiter antrainiert. Aufbau - Einheitliche Bar-Verarbeitung für paper, backtest und live; ausgetauscht werden nur Datenquelle und Broker. - Börsenanbindung über ccxt: rund 100 Börsen allein über exchange.id erreichbar. Zugangsdaten kommen über ENV-Platzhalter, der Live-Modus ist doppelt abgesichert. - Paper-Broker mit Gebühren, Slippage, Börsenpräzision und Volumengrenzen. - Online trainierte logistische Regression bewertet jedes Einstiegssignal. Sie lernt aus realen Trade-Ergebnissen, aus Shadow-Labels aller Kandidaten – auch der abgelehnten – und aus Hintergrund-Stichproben; beim Kaltstart wird sie aus der Kurshistorie vorgelernt. - Risikomanagement: Positions- und Exposure-Grenzen, ATR-Stops, Cooldown sowie Tagesverlust- und Drawdown-Notbremsen. - SQLite-Persistenz, HTTP-Status mit Prometheus-Metriken und Dashboard, Webhooks. Deployment - Containerfile (zweistufig, non-root UID 10001), podman-compose, systemd-Quadlet. - Modell und Datenbank liegen im Volume /data und überleben Neustarts. 128 Tests, ruff sauber. Verifiziert gegen echte Marktdaten sowie im gebauten Container inklusive Healthcheck und Zustandswiederherstellung.
219 lines
7.7 KiB
Python
219 lines
7.7 KiB
Python
import numpy as np
|
||
import pytest
|
||
|
||
from trademind.config import LearnerConfig, RuleConfig, StrategyConfig
|
||
from trademind.features import FEATURE_NAMES, N_FEATURES, FeatureSnapshot
|
||
from trademind.learner import AdaptiveLearner, NullLearner
|
||
from trademind.models import Action, Position
|
||
from trademind.strategy import AdaptiveStrategy, RuleStrategy, build_strategy
|
||
|
||
RULES = RuleConfig(rsi_overbought=70.0, rsi_oversold=35.0, min_holding_bars=3, trend_filter_period=100)
|
||
|
||
|
||
def snap(
|
||
*, price=100.0, rsi=50.0, rsi_prev=50.0, ema_fast=101.0, ema_slow=100.0,
|
||
ema_fast_prev=99.0, ema_slow_prev=100.0, trend_ema=95.0, atr=2.0,
|
||
) -> FeatureSnapshot:
|
||
return FeatureSnapshot(
|
||
values=np.zeros(N_FEATURES),
|
||
names=FEATURE_NAMES,
|
||
index=200,
|
||
price=price,
|
||
atr=atr,
|
||
rsi=rsi,
|
||
rsi_prev=rsi_prev,
|
||
ema_fast=ema_fast,
|
||
ema_slow=ema_slow,
|
||
ema_fast_prev=ema_fast_prev,
|
||
ema_slow_prev=ema_slow_prev,
|
||
trend_ema=trend_ema,
|
||
timestamp=1_700_000_000_000,
|
||
)
|
||
|
||
|
||
def position(bars_held: int = 10) -> Position:
|
||
return Position("BTC/USDT", 1.0, 100.0, 1_700_000_000_000, bars_held=bars_held)
|
||
|
||
|
||
# ------------------------------------------------------------------- Regeln
|
||
|
||
|
||
def test_ema_cross_up_in_uptrend_is_an_entry():
|
||
signal = RuleStrategy(RULES).evaluate("BTC/USDT", snap(), None)
|
||
assert signal.action is Action.ENTER_LONG
|
||
assert signal.reason == "ema_cross_up"
|
||
assert signal.features is not None
|
||
|
||
|
||
def test_no_entry_below_the_trend_filter():
|
||
signal = RuleStrategy(RULES).evaluate("BTC/USDT", snap(trend_ema=120.0), None)
|
||
assert signal.action is Action.HOLD
|
||
|
||
|
||
def test_no_entry_when_already_overbought():
|
||
signal = RuleStrategy(RULES).evaluate("BTC/USDT", snap(rsi=75.0), None)
|
||
assert signal.action is Action.HOLD
|
||
|
||
|
||
def test_oversold_pullback_is_an_entry():
|
||
signal = RuleStrategy(RULES).evaluate(
|
||
"BTC/USDT", snap(rsi=30.0, ema_fast_prev=101.0, ema_slow_prev=100.0), None
|
||
)
|
||
assert signal.action is Action.ENTER_LONG
|
||
assert signal.reason == "pullback_oversold"
|
||
|
||
|
||
def test_trend_filter_can_be_switched_off():
|
||
rules = RuleConfig(trend_filter_period=0)
|
||
signal = RuleStrategy(rules).evaluate("BTC/USDT", snap(trend_ema=120.0), None)
|
||
assert signal.action is Action.ENTER_LONG
|
||
|
||
|
||
def test_ema_cross_down_exits():
|
||
strategy = RuleStrategy(RULES)
|
||
s = snap(ema_fast=99.0, ema_slow=100.0, ema_fast_prev=101.0, ema_slow_prev=100.0)
|
||
assert strategy.evaluate("BTC/USDT", s, position()).action is Action.EXIT_LONG
|
||
|
||
|
||
def test_high_rsi_alone_does_not_exit():
|
||
"""Ein hoher RSI ist im Aufwärtstrend normal – er darf keinen Ausstieg auslösen."""
|
||
strategy = RuleStrategy(RULES)
|
||
s = snap(rsi=85.0, rsi_prev=80.0)
|
||
assert strategy.evaluate("BTC/USDT", s, position()).action is Action.HOLD
|
||
|
||
|
||
def test_rsi_turning_down_out_of_overbought_exits():
|
||
strategy = RuleStrategy(RULES)
|
||
s = snap(rsi=68.0, rsi_prev=74.0)
|
||
signal = strategy.evaluate("BTC/USDT", s, position())
|
||
assert signal.action is Action.EXIT_LONG
|
||
assert signal.reason == "rsi_momentum_fade"
|
||
|
||
|
||
def test_minimum_holding_period_blocks_early_signal_exits():
|
||
strategy = RuleStrategy(RULES)
|
||
s = snap(ema_fast=99.0, ema_slow=100.0, ema_fast_prev=101.0, ema_slow_prev=100.0)
|
||
assert strategy.evaluate("BTC/USDT", s, position(bars_held=1)).action is Action.HOLD
|
||
assert strategy.evaluate("BTC/USDT", s, position(bars_held=3)).action is Action.EXIT_LONG
|
||
|
||
|
||
# ----------------------------------------------------------------- Adaptive
|
||
|
||
|
||
def make_adaptive(threshold=0.55, exploration=0.0, warmup=5) -> AdaptiveStrategy:
|
||
config = StrategyConfig(
|
||
name="adaptive",
|
||
rules=RULES,
|
||
learner=LearnerConfig(
|
||
entry_threshold=threshold,
|
||
exploration_rate=exploration,
|
||
warmup_samples=warmup,
|
||
background_sample_every_n_bars=0,
|
||
),
|
||
)
|
||
return AdaptiveStrategy(config, AdaptiveLearner(config.learner, N_FEATURES, seed=1))
|
||
|
||
|
||
def test_warmup_lets_every_rule_signal_through():
|
||
strategy = make_adaptive(threshold=0.99, warmup=1_000)
|
||
signal = strategy.evaluate("BTC/USDT", snap(), None)
|
||
assert signal.action is Action.ENTER_LONG
|
||
assert "warmup" in signal.reason
|
||
|
||
|
||
def test_model_can_veto_a_rule_signal():
|
||
strategy = make_adaptive(threshold=0.99, warmup=1)
|
||
strategy.learner.observe(np.zeros(N_FEATURES), 0.0)
|
||
signal = strategy.evaluate("BTC/USDT", snap(), None)
|
||
assert signal.action is Action.HOLD
|
||
assert "abgelehnt" in signal.reason
|
||
|
||
|
||
def test_low_threshold_lets_signals_pass():
|
||
strategy = make_adaptive(threshold=0.0, warmup=1)
|
||
strategy.learner.observe(np.zeros(N_FEATURES), 0.0)
|
||
assert strategy.evaluate("BTC/USDT", snap(), None).action is Action.ENTER_LONG
|
||
|
||
|
||
def test_exploration_overrides_a_veto():
|
||
strategy = make_adaptive(threshold=0.99, exploration=1.0, warmup=1)
|
||
strategy.learner.observe(np.zeros(N_FEATURES), 0.0)
|
||
signal = strategy.evaluate("BTC/USDT", snap(), None)
|
||
assert signal.action is Action.ENTER_LONG
|
||
assert signal.exploratory is True
|
||
|
||
|
||
def test_every_candidate_is_registered_for_labelling():
|
||
"""Auch abgelehnte Signale müssen gelabelt werden – sonst lernt der Bot nichts dazu."""
|
||
strategy = make_adaptive(threshold=0.99, warmup=1)
|
||
strategy.learner.observe(np.zeros(N_FEATURES), 0.0)
|
||
strategy.evaluate("BTC/USDT", snap(), None)
|
||
assert strategy.learner.pending_count == 1
|
||
|
||
|
||
def test_no_candidate_no_pending_label():
|
||
strategy = make_adaptive()
|
||
strategy.evaluate("BTC/USDT", snap(trend_ema=120.0), None) # kein Setup
|
||
assert strategy.learner.pending_count == 0
|
||
|
||
|
||
def test_background_samples_are_registered_on_schedule():
|
||
config = StrategyConfig(
|
||
name="adaptive",
|
||
rules=RULES,
|
||
learner=LearnerConfig(background_sample_every_n_bars=5, background_sample_weight=0.5),
|
||
)
|
||
strategy = AdaptiveStrategy(config, AdaptiveLearner(config.learner, N_FEATURES, seed=1))
|
||
for bar in range(10):
|
||
strategy.on_bar("BTC/USDT", snap(), bar, 101.0, 99.0, 100.0)
|
||
assert strategy.background_samples == 2 # Bar 0 und Bar 5
|
||
|
||
|
||
def test_background_sampling_can_be_disabled():
|
||
strategy = make_adaptive() # background_sample_every_n_bars=0
|
||
for bar in range(20):
|
||
strategy.on_bar("BTC/USDT", snap(), bar, 101.0, 99.0, 100.0)
|
||
assert strategy.background_samples == 0
|
||
|
||
|
||
def test_closed_trade_feeds_the_learner():
|
||
strategy = make_adaptive()
|
||
strategy.learner.autosave = False
|
||
pos = position()
|
||
pos.entry_features = np.ones(N_FEATURES)
|
||
strategy.on_trade_closed(pos, pnl_quote=25.0)
|
||
assert strategy.learner.stats.trade_samples == 1
|
||
|
||
|
||
def test_snapshot_reports_acceptance():
|
||
strategy = make_adaptive(threshold=0.0, warmup=1)
|
||
for _ in range(3):
|
||
strategy.evaluate("BTC/USDT", snap(), None)
|
||
data = strategy.snapshot()
|
||
assert data["candidates_seen"] == 3
|
||
assert data["acceptance_rate"] == pytest.approx(1.0)
|
||
assert "learner" in data
|
||
|
||
|
||
# ------------------------------------------------------------------- Factory
|
||
|
||
|
||
def test_build_strategy_rules_variant():
|
||
strategy = build_strategy(StrategyConfig(name="rules"), N_FEATURES)
|
||
assert isinstance(strategy, RuleStrategy)
|
||
|
||
|
||
def test_build_strategy_without_learning_uses_null_learner(tmp_path):
|
||
config = StrategyConfig(
|
||
name="adaptive", learner=LearnerConfig(enabled=False, model_path=str(tmp_path / "m.npz"))
|
||
)
|
||
strategy = build_strategy(config, N_FEATURES)
|
||
assert isinstance(strategy.learner, NullLearner)
|
||
|
||
|
||
def test_build_strategy_can_skip_loading(tmp_path):
|
||
config = StrategyConfig(name="adaptive", learner=LearnerConfig(model_path=str(tmp_path / "m.npz")))
|
||
strategy = build_strategy(config, N_FEATURES, load_model=False)
|
||
assert isinstance(strategy.learner, AdaptiveLearner)
|
||
assert strategy.learner.stats.samples_seen == 0
|