Initial commit: TradeMind – Krypto-Trading-Bot mit Lernmodus
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.
This commit is contained in:
Tobias Zimmermann
@@ -0,0 +1,84 @@
|
||||
"""Gemeinsame Fixtures: synthetische Kursverläufe ohne Netzwerkzugriff."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from trademind.config import Config, PaperConfig, RiskConfig, RuleConfig
|
||||
from trademind.models import Candles
|
||||
|
||||
BAR_MS = 300_000 # 5 Minuten
|
||||
|
||||
|
||||
def make_candles(
|
||||
symbol: str = "BTC/USDT",
|
||||
n: int = 600,
|
||||
start_price: float = 30_000.0,
|
||||
trend: float = 0.0002,
|
||||
noise: float = 0.002,
|
||||
cycle: float = 0.0,
|
||||
cycle_period: int = 80,
|
||||
seed: int = 7,
|
||||
timeframe: str = "5m",
|
||||
start_ts: int = 1_700_000_000_000,
|
||||
) -> Candles:
|
||||
"""Erzeugt eine plausible OHLCV-Serie (geometrischer Random Walk plus optionale Welle)."""
|
||||
rng = np.random.default_rng(seed)
|
||||
steps = rng.normal(loc=trend, scale=noise, size=n)
|
||||
if cycle:
|
||||
steps += cycle * np.sin(2 * np.pi * np.arange(n) / cycle_period)
|
||||
close = start_price * np.exp(np.cumsum(steps))
|
||||
|
||||
open_ = np.concatenate([[start_price], close[:-1]])
|
||||
spread = np.abs(rng.normal(0.0, noise * 0.8, size=n)) * close
|
||||
high = np.maximum(open_, close) + spread
|
||||
low = np.minimum(open_, close) - spread
|
||||
low = np.maximum(low, close * 0.5)
|
||||
volume = rng.lognormal(mean=3.0, sigma=0.4, size=n)
|
||||
timestamp = start_ts + np.arange(n, dtype=np.int64) * BAR_MS
|
||||
|
||||
return Candles(
|
||||
symbol=symbol,
|
||||
timeframe=timeframe,
|
||||
timestamp=timestamp,
|
||||
open=open_,
|
||||
high=high,
|
||||
low=low,
|
||||
close=close,
|
||||
volume=volume,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def candles() -> Candles:
|
||||
return make_candles()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def rules() -> RuleConfig:
|
||||
return RuleConfig()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def paper_config() -> PaperConfig:
|
||||
return PaperConfig(starting_balance=10_000.0, fee_rate=0.001, slippage_bps=5.0)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def risk_config() -> RiskConfig:
|
||||
return RiskConfig(max_open_positions=2, max_position_pct=0.25, cooldown_bars_after_exit=0)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def base_config(tmp_path) -> Config:
|
||||
return Config.model_validate(
|
||||
{
|
||||
"mode": "backtest",
|
||||
"market": {"symbols": ["BTC/USDT"], "timeframe": "5m", "history_bars": 300},
|
||||
"paper": {"starting_balance": 10_000.0},
|
||||
"storage": {"database_path": str(tmp_path / "test.sqlite3")},
|
||||
"server": {"enabled": False},
|
||||
"strategy": {"learner": {"model_path": str(tmp_path / "model.npz"), "warmup_samples": 20}},
|
||||
}
|
||||
)
|
||||
@@ -0,0 +1,92 @@
|
||||
import pytest
|
||||
|
||||
from trademind.broker import InsufficientFunds, OrderRejected, PaperBroker
|
||||
from trademind.config import PaperConfig
|
||||
from trademind.models import Side
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def broker(paper_config) -> PaperBroker:
|
||||
return PaperBroker(paper_config)
|
||||
|
||||
|
||||
async def test_buy_applies_slippage_and_fee(broker):
|
||||
fill = await broker.execute("BTC/USDT", Side.BUY, 0.1, 30_000.0)
|
||||
assert fill.price == pytest.approx(30_000.0 * 1.0005) # 5 bps Slippage
|
||||
assert fill.fee_quote == pytest.approx(fill.notional * 0.001)
|
||||
assert await broker.cash() == pytest.approx(10_000.0 - fill.notional - fill.fee_quote)
|
||||
assert await broker.holdings("BTC/USDT") == pytest.approx(0.1)
|
||||
|
||||
|
||||
async def test_sell_applies_slippage_in_the_other_direction(broker):
|
||||
await broker.execute("BTC/USDT", Side.BUY, 0.1, 30_000.0)
|
||||
fill = await broker.execute("BTC/USDT", Side.SELL, 0.1, 31_000.0)
|
||||
assert fill.price == pytest.approx(31_000.0 * 0.9995)
|
||||
assert await broker.holdings("BTC/USDT") == 0.0
|
||||
|
||||
|
||||
async def test_round_trip_at_constant_price_loses_exactly_the_costs(broker):
|
||||
price = 30_000.0
|
||||
await broker.execute("BTC/USDT", Side.BUY, 0.1, price)
|
||||
await broker.execute("BTC/USDT", Side.SELL, 0.1, price)
|
||||
# 2 × 0,1 % Gebühr + 2 × 5 bps Slippage auf ein Volumen von ~3000
|
||||
assert await broker.cash() == pytest.approx(10_000.0 - 9.0, abs=0.2)
|
||||
|
||||
|
||||
async def test_buy_is_scaled_down_to_available_cash(broker):
|
||||
fill = await broker.execute("BTC/USDT", Side.BUY, 10.0, 30_000.0) # 300k gewünscht
|
||||
assert fill.amount < 10.0
|
||||
assert await broker.cash() == pytest.approx(0.0, abs=1e-6)
|
||||
|
||||
|
||||
async def test_buy_without_any_cash_raises():
|
||||
broker = PaperBroker(PaperConfig(starting_balance=1.0, fee_rate=0.001))
|
||||
broker.market_info = {"BTC/USDT": {"amount_precision": 4}}
|
||||
with pytest.raises(InsufficientFunds):
|
||||
await broker.execute("BTC/USDT", Side.BUY, 1.0, 30_000.0)
|
||||
|
||||
|
||||
async def test_sell_without_position_raises(broker):
|
||||
with pytest.raises(OrderRejected):
|
||||
await broker.execute("BTC/USDT", Side.SELL, 1.0, 30_000.0)
|
||||
|
||||
|
||||
async def test_sell_is_capped_at_held_amount(broker):
|
||||
await broker.execute("BTC/USDT", Side.BUY, 0.1, 30_000.0)
|
||||
fill = await broker.execute("BTC/USDT", Side.SELL, 5.0, 30_000.0)
|
||||
assert fill.amount == pytest.approx(0.1)
|
||||
|
||||
|
||||
async def test_volume_participation_caps_the_order(broker):
|
||||
fill = await broker.execute("BTC/USDT", Side.BUY, 0.1, 30_000.0, bar_volume=0.2)
|
||||
assert fill.amount == pytest.approx(0.02) # 10 % von 0,2
|
||||
assert fill.requested_amount == pytest.approx(0.1)
|
||||
|
||||
|
||||
async def test_amount_precision_rounds_down(broker):
|
||||
broker.market_info = {"BTC/USDT": {"amount_precision": 3}}
|
||||
fill = await broker.execute("BTC/USDT", Side.BUY, 0.123456, 30_000.0)
|
||||
assert fill.amount == pytest.approx(0.123)
|
||||
|
||||
|
||||
async def test_step_size_precision_is_supported(broker):
|
||||
broker.market_info = {"BTC/USDT": {"amount_precision": 0.05}}
|
||||
fill = await broker.execute("BTC/USDT", Side.BUY, 0.17, 100.0)
|
||||
assert fill.amount == pytest.approx(0.15)
|
||||
|
||||
|
||||
async def test_invalid_inputs_are_rejected(broker):
|
||||
with pytest.raises(OrderRejected):
|
||||
await broker.execute("BTC/USDT", Side.BUY, 0.0, 30_000.0)
|
||||
with pytest.raises(OrderRejected):
|
||||
await broker.execute("BTC/USDT", Side.BUY, 1.0, 0.0)
|
||||
|
||||
|
||||
async def test_state_round_trip(broker):
|
||||
await broker.execute("BTC/USDT", Side.BUY, 0.05, 30_000.0)
|
||||
state = broker.state()
|
||||
|
||||
restored = PaperBroker(broker.config)
|
||||
restored.restore(state["cash"], state["holdings"], state["total_fees"])
|
||||
assert await restored.cash() == pytest.approx(await broker.cash())
|
||||
assert await restored.holdings("BTC/USDT") == pytest.approx(0.05)
|
||||
@@ -0,0 +1,96 @@
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from trademind.config import LIVE_CONFIRMATION_PHRASE, Config, Mode, load_config
|
||||
|
||||
MINIMAL = """
|
||||
mode: paper
|
||||
market:
|
||||
symbols: [btc/usdt]
|
||||
"""
|
||||
|
||||
|
||||
def write(tmp_path, text: str):
|
||||
path = tmp_path / "config.yaml"
|
||||
path.write_text(text, encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
def test_minimal_config_uses_defaults(tmp_path):
|
||||
config = load_config(write(tmp_path, MINIMAL))
|
||||
assert config.mode is Mode.PAPER
|
||||
assert config.market.symbols == ["BTC/USDT"] # normalisiert
|
||||
assert config.exchange.id == "binance"
|
||||
assert config.risk.max_open_positions == 3
|
||||
|
||||
|
||||
def test_env_placeholders_are_resolved(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("MY_KEY", "abc123")
|
||||
config = load_config(
|
||||
write(tmp_path, "mode: paper\nexchange:\n api_key: ${MY_KEY}\n api_secret: ${MISSING}\n")
|
||||
)
|
||||
assert config.exchange.api_key == "abc123"
|
||||
assert config.exchange.api_secret is None
|
||||
|
||||
|
||||
def test_env_placeholder_default_value(tmp_path, monkeypatch):
|
||||
monkeypatch.delenv("TRADEMIND_EXCHANGE", raising=False)
|
||||
config = load_config(write(tmp_path, "exchange:\n id: ${TRADEMIND_EXCHANGE:-kraken}\n"))
|
||||
assert config.exchange.id == "kraken"
|
||||
|
||||
|
||||
def test_double_underscore_env_override(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("TRADEMIND__RISK__MAX_OPEN_POSITIONS", "7")
|
||||
monkeypatch.setenv("TRADEMIND__MARKET__SYMBOLS", "BTC/USDT,ETH/USDT")
|
||||
monkeypatch.setenv("TRADEMIND__EXCHANGE__SANDBOX", "false")
|
||||
config = load_config(write(tmp_path, MINIMAL))
|
||||
assert config.risk.max_open_positions == 7
|
||||
assert config.market.symbols == ["BTC/USDT", "ETH/USDT"]
|
||||
assert config.exchange.sandbox is False
|
||||
|
||||
|
||||
def test_live_mode_requires_confirmation(tmp_path):
|
||||
text = "mode: live\nexchange:\n api_key: k\n api_secret: s\n"
|
||||
with pytest.raises(ValueError, match="live_confirmation"):
|
||||
load_config(write(tmp_path, text))
|
||||
|
||||
|
||||
def test_live_mode_requires_credentials(tmp_path):
|
||||
text = f"mode: live\nlive_confirmation: {LIVE_CONFIRMATION_PHRASE}\n"
|
||||
with pytest.raises(ValueError, match="api_key"):
|
||||
load_config(write(tmp_path, text))
|
||||
|
||||
|
||||
def test_live_mode_accepted_when_complete(tmp_path):
|
||||
text = (
|
||||
f"mode: live\nlive_confirmation: {LIVE_CONFIRMATION_PHRASE}\n"
|
||||
"exchange:\n api_key: k\n api_secret: s\n"
|
||||
)
|
||||
config = load_config(write(tmp_path, text))
|
||||
assert config.mode is Mode.LIVE
|
||||
assert config.is_simulated is False
|
||||
|
||||
|
||||
def test_unknown_key_is_rejected(tmp_path):
|
||||
with pytest.raises(ValidationError):
|
||||
load_config(write(tmp_path, "mode: paper\nrisk:\n typo_here: 5\n"))
|
||||
|
||||
|
||||
def test_ema_periods_must_be_ordered(tmp_path):
|
||||
with pytest.raises(ValueError, match="fast_ema"):
|
||||
load_config(write(tmp_path, "strategy:\n rules:\n fast_ema: 30\n slow_ema: 10\n"))
|
||||
|
||||
|
||||
def test_missing_file_raises(tmp_path):
|
||||
with pytest.raises(FileNotFoundError):
|
||||
load_config(tmp_path / "nope.yaml")
|
||||
|
||||
|
||||
def test_empty_symbols_rejected(tmp_path):
|
||||
with pytest.raises(ValidationError):
|
||||
load_config(write(tmp_path, "market:\n symbols: []\n"))
|
||||
|
||||
|
||||
def test_config_is_simulated_flag():
|
||||
assert Config(mode=Mode.PAPER).is_simulated
|
||||
assert Config(mode=Mode.BACKTEST).is_simulated
|
||||
@@ -0,0 +1,334 @@
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from trademind.backtest import BacktestRunner, trades_csv
|
||||
from trademind.broker import PaperBroker
|
||||
from trademind.config import Config
|
||||
from trademind.data import DataFeed
|
||||
from trademind.engine import TradingEngine
|
||||
from trademind.features import N_FEATURES
|
||||
from trademind.models import Candles, ExitReason, Side
|
||||
from trademind.portfolio import Portfolio
|
||||
from trademind.risk import RiskManager
|
||||
from trademind.storage import NullStorage, Storage
|
||||
from trademind.strategy import build_strategy
|
||||
|
||||
from .conftest import make_candles
|
||||
|
||||
|
||||
class StaticFeed(DataFeed):
|
||||
"""Liefert immer dasselbe Fenster – für Tests, die den Feed nicht brauchen."""
|
||||
|
||||
def __init__(self, candles: Candles | None = None) -> None:
|
||||
self.candles = candles
|
||||
|
||||
async def fetch(self, symbol: str, timeframe: str, limit: int) -> Candles:
|
||||
if self.candles is None:
|
||||
raise AssertionError("Feed sollte in diesem Test nicht abgefragt werden")
|
||||
return self.candles
|
||||
|
||||
|
||||
class GrowingFeed(DataFeed):
|
||||
"""Gibt bei jedem Abruf ein um eine Kerze längeres Fenster zurück (simuliert Live-Betrieb)."""
|
||||
|
||||
def __init__(self, candles: Candles, start: int) -> None:
|
||||
self.full = candles
|
||||
self.cursor = start
|
||||
self.calls = 0
|
||||
|
||||
async def fetch(self, symbol: str, timeframe: str, limit: int) -> Candles:
|
||||
self.calls += 1
|
||||
stop = min(self.cursor, len(self.full))
|
||||
start = max(0, stop - limit)
|
||||
return self.full.slice(start, stop)
|
||||
|
||||
def advance(self) -> None:
|
||||
self.cursor += 1
|
||||
|
||||
|
||||
def build_engine(config: Config, feed: DataFeed | None = None, storage=None) -> TradingEngine:
|
||||
broker = PaperBroker(config.paper)
|
||||
strategy = build_strategy(config.strategy, N_FEATURES, seed=3, load_model=False)
|
||||
learner = getattr(strategy, "learner", None)
|
||||
if learner is not None:
|
||||
learner.autosave = False
|
||||
return TradingEngine(
|
||||
config=config,
|
||||
broker=broker,
|
||||
feed=feed or StaticFeed(),
|
||||
strategy=strategy,
|
||||
portfolio=Portfolio(config.paper.starting_balance, config.paper.quote_currency),
|
||||
risk=RiskManager(config.risk),
|
||||
storage=storage or NullStorage(),
|
||||
)
|
||||
|
||||
|
||||
def cyclical_series(symbol: str = "BTC/USDT", n: int = 900, seed: int = 4) -> Candles:
|
||||
"""Schwingender Verlauf – erzeugt zuverlässig Ein- und Ausstiegssignale."""
|
||||
return make_candles(symbol=symbol, n=n, trend=0.0001, noise=0.0015, cycle=0.0025,
|
||||
cycle_period=60, seed=seed)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ Backtest
|
||||
|
||||
|
||||
async def test_backtest_runs_and_produces_trades(base_config):
|
||||
engine = build_engine(base_config)
|
||||
await engine.prepare()
|
||||
report = await BacktestRunner(engine, {"BTC/USDT": cyclical_series()}, progress_every=0).run()
|
||||
|
||||
assert report.bars > 0
|
||||
assert report.portfolio["trades"] > 0
|
||||
assert engine.portfolio.positions == {} # am Ende glattgestellt
|
||||
assert report.portfolio["equity"] > 0
|
||||
|
||||
|
||||
async def test_cash_and_equity_stay_consistent(base_config):
|
||||
engine = build_engine(base_config)
|
||||
await engine.prepare()
|
||||
await BacktestRunner(engine, {"BTC/USDT": cyclical_series()}, progress_every=0).run()
|
||||
|
||||
cash = await engine.broker.cash()
|
||||
realized = sum(t.pnl_quote for t in engine.portfolio.trades)
|
||||
assert cash == pytest.approx(base_config.paper.starting_balance + realized, abs=1e-6)
|
||||
assert cash >= 0.0
|
||||
|
||||
|
||||
async def test_stop_loss_bounds_the_worst_trade(base_config):
|
||||
config = Config.model_validate(
|
||||
{**base_config.model_dump(), "risk": {**base_config.risk.model_dump(),
|
||||
"stop_loss_atr_mult": 1.0,
|
||||
"take_profit_atr_mult": 10.0}}
|
||||
)
|
||||
engine = build_engine(config)
|
||||
await engine.prepare()
|
||||
await BacktestRunner(engine, {"BTC/USDT": cyclical_series()}, progress_every=0).run()
|
||||
|
||||
stopped = [t for t in engine.portfolio.trades if t.exit_reason is ExitReason.STOP_LOSS]
|
||||
assert stopped, "Bei engem Stop sollten Stop-Ausstiege vorkommen"
|
||||
for trade in stopped:
|
||||
assert trade.pnl_pct > -0.25 # ein Stop begrenzt den Verlust deutlich
|
||||
|
||||
|
||||
async def test_position_limit_is_never_exceeded(base_config):
|
||||
config = Config.model_validate(
|
||||
{
|
||||
**base_config.model_dump(),
|
||||
"market": {**base_config.market.model_dump(), "symbols": ["BTC/USDT", "ETH/USDT", "SOL/USDT"]},
|
||||
"risk": {**base_config.risk.model_dump(), "max_open_positions": 2},
|
||||
}
|
||||
)
|
||||
engine = build_engine(config)
|
||||
await engine.prepare()
|
||||
|
||||
observed_max = 0
|
||||
original = engine.process_bar
|
||||
|
||||
async def spy(symbol, snapshot, bar):
|
||||
nonlocal observed_max
|
||||
await original(symbol, snapshot, bar)
|
||||
observed_max = max(observed_max, len(engine.portfolio.positions))
|
||||
|
||||
engine.process_bar = spy # type: ignore[method-assign]
|
||||
series = {
|
||||
"BTC/USDT": cyclical_series("BTC/USDT", seed=4),
|
||||
"ETH/USDT": cyclical_series("ETH/USDT", seed=5),
|
||||
"SOL/USDT": cyclical_series("SOL/USDT", seed=6),
|
||||
}
|
||||
await BacktestRunner(engine, series, progress_every=0).run()
|
||||
assert observed_max <= 2
|
||||
|
||||
|
||||
async def test_learner_collects_samples_during_a_backtest(base_config):
|
||||
engine = build_engine(base_config)
|
||||
await engine.prepare()
|
||||
await BacktestRunner(engine, {"BTC/USDT": cyclical_series()}, progress_every=0).run()
|
||||
|
||||
learner = engine.strategy.learner
|
||||
assert learner.stats.samples_seen > 0
|
||||
assert learner.stats.trade_samples > 0 # aus echten Trades gelernt
|
||||
assert learner.stats.shadow_samples > 0 # und aus nicht gehandelten Signalen
|
||||
assert learner.stats.updates > 0
|
||||
|
||||
|
||||
async def test_rules_strategy_needs_no_learner(base_config):
|
||||
config = Config.model_validate({**base_config.model_dump(), "strategy": {"name": "rules"}})
|
||||
engine = build_engine(config)
|
||||
await engine.prepare()
|
||||
report = await BacktestRunner(engine, {"BTC/USDT": cyclical_series()}, progress_every=0).run()
|
||||
assert report.strategy == {"strategy": "rules"}
|
||||
|
||||
|
||||
async def test_higher_threshold_trades_less(base_config):
|
||||
def run_config(threshold: float) -> Config:
|
||||
learner = {**base_config.strategy.learner.model_dump(),
|
||||
"entry_threshold": threshold, "exploration_rate": 0.0, "warmup_samples": 30}
|
||||
return Config.model_validate(
|
||||
{**base_config.model_dump(),
|
||||
"strategy": {**base_config.strategy.model_dump(), "learner": learner}}
|
||||
)
|
||||
|
||||
rates = []
|
||||
for threshold in (0.0, 0.95):
|
||||
engine = build_engine(run_config(threshold))
|
||||
await engine.prepare()
|
||||
await BacktestRunner(engine, {"BTC/USDT": cyclical_series()}, progress_every=0).run()
|
||||
assert engine.strategy.candidates_seen >= 10, "zu wenige Signale für einen Vergleich"
|
||||
rates.append(engine.strategy.candidates_accepted / engine.strategy.candidates_seen)
|
||||
|
||||
assert rates[0] == pytest.approx(1.0) # Schwelle 0 lässt alles durch
|
||||
assert rates[1] < rates[0] # hohe Schwelle filtert
|
||||
|
||||
|
||||
async def test_trades_csv_has_one_row_per_trade(base_config):
|
||||
engine = build_engine(base_config)
|
||||
await engine.prepare()
|
||||
await BacktestRunner(engine, {"BTC/USDT": cyclical_series()}, progress_every=0).run()
|
||||
lines = trades_csv(engine).strip().splitlines()
|
||||
assert len(lines) == len(engine.portfolio.trades) + 1
|
||||
|
||||
|
||||
async def test_short_series_is_rejected(base_config):
|
||||
engine = build_engine(base_config)
|
||||
await engine.prepare()
|
||||
with pytest.raises(ValueError, match="genug Kerzen"):
|
||||
await BacktestRunner(engine, {"BTC/USDT": make_candles(n=50)}, progress_every=0).run()
|
||||
|
||||
|
||||
# ----------------------------------------------------------- Live-artiger Loop
|
||||
|
||||
|
||||
async def test_tick_only_acts_on_new_bars(base_config):
|
||||
series = cyclical_series(n=400)
|
||||
feed = GrowingFeed(series, start=300)
|
||||
engine = build_engine(base_config, feed=feed)
|
||||
await engine.prepare()
|
||||
|
||||
await engine._tick(300)
|
||||
first = dict(engine.bar_counter)
|
||||
await engine._tick(300) # keine neue Kerze
|
||||
assert engine.bar_counter == first
|
||||
|
||||
feed.advance()
|
||||
await engine._tick(300)
|
||||
assert engine.bar_counter["BTC/USDT"] == first["BTC/USDT"] + 1
|
||||
|
||||
|
||||
async def test_bootstrap_trains_the_model_from_history(base_config):
|
||||
"""Ein Kaltstart muss das Modell aus der Historie vorlernen, nicht tagelang warten."""
|
||||
series = cyclical_series(n=900)
|
||||
engine = build_engine(base_config, feed=GrowingFeed(series, start=900))
|
||||
await engine.prepare()
|
||||
assert engine.strategy.learner.ready is False
|
||||
|
||||
await engine.bootstrap_learner()
|
||||
|
||||
learner = engine.strategy.learner
|
||||
assert learner.stats.samples_seen > base_config.strategy.learner.warmup_samples
|
||||
assert learner.ready is True
|
||||
assert engine.portfolio.trades == [] # Vorlernen handelt nicht
|
||||
assert engine.bar_counter["BTC/USDT"] > 0 # Zähler schließt an die Historie an
|
||||
assert engine.last_bar_ts["BTC/USDT"] == int(series.timestamp[engine.bar_counter["BTC/USDT"]])
|
||||
|
||||
|
||||
async def test_bootstrap_is_skipped_for_a_trained_model(base_config):
|
||||
engine = build_engine(base_config, feed=GrowingFeed(cyclical_series(n=900), start=900))
|
||||
await engine.prepare()
|
||||
for _ in range(base_config.strategy.learner.warmup_samples):
|
||||
engine.strategy.learner.observe(np.zeros(N_FEATURES), 1.0)
|
||||
seen = engine.strategy.learner.stats.samples_seen
|
||||
|
||||
await engine.bootstrap_learner()
|
||||
assert engine.strategy.learner.stats.samples_seen == seen
|
||||
|
||||
|
||||
async def test_bootstrap_survives_a_broken_feed(base_config):
|
||||
class BrokenFeed(DataFeed):
|
||||
async def fetch(self, symbol, timeframe, limit):
|
||||
raise RuntimeError("Börse nicht erreichbar")
|
||||
|
||||
engine = build_engine(base_config, feed=BrokenFeed())
|
||||
await engine.prepare()
|
||||
await engine.bootstrap_learner() # darf nicht werfen
|
||||
assert engine.strategy.learner.ready is False
|
||||
|
||||
|
||||
async def test_tick_waits_for_enough_history(base_config):
|
||||
feed = GrowingFeed(cyclical_series(n=400), start=60)
|
||||
engine = build_engine(base_config, feed=feed)
|
||||
await engine.prepare()
|
||||
await engine._tick(300)
|
||||
assert engine.bar_counter["BTC/USDT"] == 0 # Indikatoren noch nicht warm
|
||||
|
||||
|
||||
# --------------------------------------------------------------- Persistenz
|
||||
|
||||
|
||||
async def test_state_survives_a_restart(base_config, tmp_path):
|
||||
storage = Storage(tmp_path / "state.sqlite3")
|
||||
config = Config.model_validate({**base_config.model_dump(), "mode": "paper"})
|
||||
engine = build_engine(config, feed=StaticFeed(), storage=storage)
|
||||
await engine.prepare()
|
||||
|
||||
# Position künstlich eröffnen und Zustand sichern.
|
||||
fill = await engine.broker.execute("BTC/USDT", Side.BUY, 0.05, 30_000.0)
|
||||
engine.portfolio.open_position(
|
||||
fill, stop_loss=29_000.0, take_profit=32_000.0,
|
||||
features=np.ones(N_FEATURES), confidence=0.7, exploratory=False,
|
||||
)
|
||||
engine._cash = await engine.broker.cash()
|
||||
engine._persist_state()
|
||||
|
||||
revived = build_engine(config, feed=StaticFeed(), storage=storage)
|
||||
await revived.prepare()
|
||||
|
||||
assert "BTC/USDT" in revived.portfolio.positions
|
||||
restored = revived.portfolio.positions["BTC/USDT"]
|
||||
assert restored.amount == pytest.approx(0.05)
|
||||
assert restored.stop_loss == pytest.approx(29_000.0)
|
||||
assert restored.entry_features is not None
|
||||
assert await revived.broker.cash() == pytest.approx(await engine.broker.cash())
|
||||
storage.close()
|
||||
|
||||
|
||||
async def test_state_from_another_mode_is_ignored(base_config, tmp_path):
|
||||
storage = Storage(tmp_path / "state.sqlite3")
|
||||
paper = Config.model_validate({**base_config.model_dump(), "mode": "paper"})
|
||||
engine = build_engine(paper, storage=storage)
|
||||
await engine.prepare()
|
||||
engine._persist_state()
|
||||
|
||||
backtest = Config.model_validate({**base_config.model_dump(), "mode": "backtest"})
|
||||
other = build_engine(backtest, storage=storage)
|
||||
await other.prepare()
|
||||
assert other.portfolio.positions == {}
|
||||
storage.close()
|
||||
|
||||
|
||||
async def test_trades_are_written_to_the_database(base_config, tmp_path):
|
||||
storage = Storage(tmp_path / "trades.sqlite3")
|
||||
engine = build_engine(base_config, storage=storage)
|
||||
await engine.prepare()
|
||||
await BacktestRunner(engine, {"BTC/USDT": cyclical_series()}, progress_every=0).run()
|
||||
|
||||
assert storage.trade_count() == len(engine.portfolio.trades)
|
||||
per_symbol = storage.performance_by_symbol()
|
||||
assert per_symbol and per_symbol[0]["symbol"] == "BTC/USDT"
|
||||
storage.close()
|
||||
|
||||
|
||||
# ------------------------------------------------------------------- Status
|
||||
|
||||
|
||||
async def test_status_payload_is_serialisable(base_config):
|
||||
engine = build_engine(base_config)
|
||||
await engine.prepare()
|
||||
await BacktestRunner(engine, {"BTC/USDT": cyclical_series(n=400)}, progress_every=0).run()
|
||||
|
||||
import json
|
||||
|
||||
status = engine.status()
|
||||
json.dumps(status, default=str) # darf nicht werfen
|
||||
assert status["mode"] == "backtest"
|
||||
assert "portfolio" in status and "strategy" in status
|
||||
assert len(status["feature_weights"]) == N_FEATURES
|
||||
@@ -0,0 +1,83 @@
|
||||
import numpy as np
|
||||
|
||||
from trademind.features import (
|
||||
FEATURE_NAMES,
|
||||
N_FEATURES,
|
||||
build_feature_matrix,
|
||||
compute_features,
|
||||
required_bars,
|
||||
)
|
||||
|
||||
from .conftest import make_candles
|
||||
|
||||
|
||||
def test_matrix_has_expected_shape(candles, rules):
|
||||
matrix = build_feature_matrix(candles, rules)
|
||||
assert matrix is not None
|
||||
assert matrix.values.shape == (len(candles), N_FEATURES)
|
||||
assert matrix.first_valid == required_bars(rules) - 1
|
||||
|
||||
|
||||
def test_all_feature_values_are_finite_and_bounded(candles, rules):
|
||||
matrix = build_feature_matrix(candles, rules)
|
||||
valid = matrix.values[matrix.first_valid :]
|
||||
assert np.isfinite(valid).all()
|
||||
assert np.abs(valid).max() <= 8.0
|
||||
|
||||
|
||||
def test_too_short_history_returns_none(rules):
|
||||
short = make_candles(n=50)
|
||||
assert build_feature_matrix(short, rules) is None
|
||||
assert compute_features(short, rules) is None
|
||||
|
||||
|
||||
def test_snapshot_exposes_raw_indicators(candles, rules):
|
||||
snapshot = compute_features(candles, rules)
|
||||
assert snapshot is not None
|
||||
assert snapshot.price == float(candles.close[-1])
|
||||
assert snapshot.atr > 0
|
||||
assert 0.0 <= snapshot.rsi <= 100.0
|
||||
assert len(snapshot.values) == len(FEATURE_NAMES)
|
||||
assert set(snapshot.as_dict()) == set(FEATURE_NAMES)
|
||||
|
||||
|
||||
def test_snapshot_before_warmup_is_none(candles, rules):
|
||||
matrix = build_feature_matrix(candles, rules)
|
||||
assert matrix.snapshot(matrix.first_valid - 1) is None
|
||||
assert matrix.snapshot(matrix.first_valid) is not None
|
||||
|
||||
|
||||
def test_uptrend_produces_positive_trend_distance(rules):
|
||||
up = make_candles(n=500, trend=0.001, noise=0.0005, seed=11)
|
||||
snapshot = compute_features(up, rules)
|
||||
features = snapshot.as_dict()
|
||||
assert features["trend_dist"] > 0
|
||||
assert features["ema_spread"] > 0
|
||||
|
||||
|
||||
def test_downtrend_produces_negative_trend_distance(rules):
|
||||
down = make_candles(n=500, trend=-0.001, noise=0.0005, seed=12)
|
||||
features = compute_features(down, rules).as_dict()
|
||||
assert features["trend_dist"] < 0
|
||||
assert features["ema_spread"] < 0
|
||||
|
||||
|
||||
def test_features_are_scale_invariant(rules):
|
||||
"""Ein zehnfach höherer Kurs darf die normierten Merkmale kaum verändern."""
|
||||
cheap = make_candles(n=400, start_price=100.0, seed=5)
|
||||
expensive = make_candles(n=400, start_price=1_000.0, seed=5)
|
||||
a = compute_features(cheap, rules).values
|
||||
b = compute_features(expensive, rules).values
|
||||
assert np.allclose(a, b, atol=1e-8)
|
||||
|
||||
|
||||
def test_time_features_are_on_the_unit_circle(candles, rules):
|
||||
snapshot = compute_features(candles, rules).as_dict()
|
||||
radius = snapshot["time_sin"] ** 2 + snapshot["time_cos"] ** 2
|
||||
assert radius == 1.0 or abs(radius - 1.0) < 1e-9
|
||||
|
||||
|
||||
def test_matrix_rows_match_pointwise_snapshots(candles, rules):
|
||||
matrix = build_feature_matrix(candles, rules)
|
||||
index = matrix.first_valid + 25
|
||||
assert np.allclose(matrix.snapshot(index).values, matrix.values[index])
|
||||
@@ -0,0 +1,92 @@
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from trademind.indicators import atr, bollinger, donchian_position, ema, macd, roc, rsi, sma, true_range
|
||||
|
||||
|
||||
def test_ema_of_constant_series_is_constant():
|
||||
values = np.full(50, 42.0)
|
||||
assert np.allclose(ema(values, 10), 42.0)
|
||||
|
||||
|
||||
def test_ema_reacts_faster_than_sma():
|
||||
values = np.concatenate([np.full(30, 100.0), np.full(30, 110.0)])
|
||||
fast = ema(values, 10)[35]
|
||||
slow = sma(values, 10)[35]
|
||||
assert fast > slow # EMA hat den Sprung stärker eingepreist
|
||||
|
||||
|
||||
def test_sma_matches_manual_mean():
|
||||
values = np.arange(1.0, 11.0)
|
||||
result = sma(values, 3)
|
||||
assert np.isnan(result[:2]).all()
|
||||
assert result[2] == pytest.approx(2.0)
|
||||
assert result[-1] == pytest.approx(9.0)
|
||||
|
||||
|
||||
def test_rsi_bounds_and_extremes():
|
||||
rising = np.arange(1.0, 60.0)
|
||||
values = rsi(rising, 14)
|
||||
finite = values[np.isfinite(values)]
|
||||
assert finite.min() >= 0.0 and finite.max() <= 100.0
|
||||
assert finite[-1] == pytest.approx(100.0) # nur Gewinne
|
||||
|
||||
falling = rising[::-1].copy()
|
||||
assert rsi(falling, 14)[-1] == pytest.approx(0.0)
|
||||
|
||||
|
||||
def test_rsi_of_flat_series_is_neutral():
|
||||
values = rsi(np.full(60, 25.0), 14)
|
||||
assert values[-1] == pytest.approx(50.0)
|
||||
|
||||
|
||||
def test_true_range_covers_gaps():
|
||||
high = np.array([10.0, 20.0])
|
||||
low = np.array([9.0, 19.0])
|
||||
close = np.array([9.5, 19.5])
|
||||
tr = true_range(high, low, close)
|
||||
assert tr[0] == pytest.approx(1.0)
|
||||
assert tr[1] == pytest.approx(10.5) # Lücke gegenüber dem Vortagesschluss
|
||||
|
||||
|
||||
def test_atr_is_positive_and_warm():
|
||||
n = 100
|
||||
rng = np.random.default_rng(3)
|
||||
close = 100 + np.cumsum(rng.normal(0, 1, n))
|
||||
high = close + 1.0
|
||||
low = close - 1.0
|
||||
values = atr(high, low, close, 14)
|
||||
assert np.isnan(values[:13]).all()
|
||||
assert (values[13:] > 0).all()
|
||||
|
||||
|
||||
def test_macd_histogram_is_difference():
|
||||
values = 100 + np.cumsum(np.random.default_rng(1).normal(0, 1, 200))
|
||||
line, signal, hist = macd(values)
|
||||
assert np.allclose(hist, line - signal)
|
||||
|
||||
|
||||
def test_bollinger_bands_are_ordered():
|
||||
values = 100 + np.cumsum(np.random.default_rng(2).normal(0, 1, 200))
|
||||
lower, mid, upper = bollinger(values, 20, 2.0)
|
||||
valid = ~np.isnan(mid)
|
||||
assert (lower[valid] <= mid[valid]).all()
|
||||
assert (mid[valid] <= upper[valid]).all()
|
||||
|
||||
|
||||
def test_roc_is_relative_change():
|
||||
values = np.array([100.0] * 10 + [110.0])
|
||||
assert roc(values, 10)[-1] == pytest.approx(0.10)
|
||||
|
||||
|
||||
def test_donchian_position_hits_extremes():
|
||||
close = np.array([float(i) for i in range(1, 41)])
|
||||
high = close + 0.0
|
||||
low = close - 0.0
|
||||
pos = donchian_position(high, low, close, 20)
|
||||
assert pos[-1] == pytest.approx(1.0) # Schluss auf dem Hoch der Range
|
||||
|
||||
|
||||
def test_indicators_reject_wrong_dimensions():
|
||||
with pytest.raises(ValueError):
|
||||
ema(np.zeros((5, 2)), 3)
|
||||
@@ -0,0 +1,237 @@
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from trademind.config import LearnerConfig
|
||||
from trademind.learner import (
|
||||
AdaptiveLearner,
|
||||
NullLearner,
|
||||
OnlineLogisticRegression,
|
||||
ReplayBuffer,
|
||||
RunningScaler,
|
||||
sigmoid,
|
||||
)
|
||||
|
||||
|
||||
def make_learner(tmp_path, **overrides) -> AdaptiveLearner:
|
||||
config = LearnerConfig(
|
||||
model_path=str(tmp_path / "model.npz"),
|
||||
warmup_samples=overrides.pop("warmup_samples", 20),
|
||||
batch_size=overrides.pop("batch_size", 32),
|
||||
train_every_n_samples=overrides.pop("train_every_n_samples", 1),
|
||||
learning_rate=overrides.pop("learning_rate", 0.05),
|
||||
**overrides,
|
||||
)
|
||||
return AdaptiveLearner(config, n_features=4, seed=1)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- Bausteine
|
||||
|
||||
|
||||
def test_sigmoid_is_bounded_and_stable():
|
||||
assert sigmoid(0.0) == pytest.approx(0.5)
|
||||
assert 0.0 < float(sigmoid(-1000.0)) < 1e-10
|
||||
assert float(sigmoid(1000.0)) > 1 - 1e-10
|
||||
|
||||
|
||||
def test_running_scaler_matches_numpy():
|
||||
rng = np.random.default_rng(0)
|
||||
data = rng.normal(5.0, 3.0, size=(500, 4))
|
||||
scaler = RunningScaler(4)
|
||||
for row in data:
|
||||
scaler.update(row)
|
||||
assert np.allclose(scaler.mean, data.mean(axis=0), atol=1e-9)
|
||||
assert np.allclose(scaler.std, data.std(axis=0, ddof=1), atol=1e-9)
|
||||
|
||||
|
||||
def test_scaler_clips_outliers():
|
||||
scaler = RunningScaler(2)
|
||||
for value in np.random.default_rng(1).normal(0, 1, size=(200, 2)):
|
||||
scaler.update(value)
|
||||
scaled = scaler.transform(np.array([[1e6, -1e6]]))
|
||||
assert np.abs(scaled).max() <= 6.0
|
||||
|
||||
|
||||
def test_replay_buffer_is_a_ring():
|
||||
buffer = ReplayBuffer(3, 2, np.random.default_rng(0))
|
||||
for i in range(5):
|
||||
buffer.add(np.array([i, i]), float(i % 2), 1.0)
|
||||
assert len(buffer) == 3
|
||||
x, y, w = buffer.sample(3)
|
||||
assert x.shape == (3, 2)
|
||||
assert set(np.unique(x[:, 0])).issubset({2.0, 3.0, 4.0}) # nur die letzten drei
|
||||
|
||||
|
||||
def test_logistic_regression_learns_a_separable_problem():
|
||||
rng = np.random.default_rng(0)
|
||||
model = OnlineLogisticRegression(2, learning_rate=0.1)
|
||||
x = rng.normal(0, 1, size=(400, 2))
|
||||
y = (x[:, 0] + x[:, 1] > 0).astype(float)
|
||||
for _ in range(60):
|
||||
model.partial_fit(x, y)
|
||||
predictions = model.predict_proba(x) >= 0.5
|
||||
assert (predictions == (y > 0.5)).mean() > 0.9
|
||||
|
||||
|
||||
# -------------------------------------------------------------- Lernverhalten
|
||||
|
||||
|
||||
def test_learner_is_not_ready_before_warmup(tmp_path):
|
||||
learner = make_learner(tmp_path, warmup_samples=10)
|
||||
assert learner.ready is False
|
||||
for _ in range(10):
|
||||
learner.observe(np.zeros(4), 1.0)
|
||||
assert learner.ready is True
|
||||
|
||||
|
||||
def test_learner_separates_good_from_bad_setups(tmp_path):
|
||||
"""Feature 0 entscheidet über den Ausgang – das muss das Modell finden."""
|
||||
learner = make_learner(tmp_path, warmup_samples=10, learning_rate=0.1)
|
||||
rng = np.random.default_rng(3)
|
||||
for _ in range(800):
|
||||
good = rng.random() < 0.5
|
||||
features = np.array([1.0 if good else -1.0, *rng.normal(0, 0.5, 3)])
|
||||
learner.observe(features, 1.0 if good else 0.0)
|
||||
|
||||
good_score = learner.score(np.array([1.0, 0.0, 0.0, 0.0]))
|
||||
bad_score = learner.score(np.array([-1.0, 0.0, 0.0, 0.0]))
|
||||
assert good_score > 0.7
|
||||
assert bad_score < 0.3
|
||||
assert learner.stats.accuracy > 0.8
|
||||
|
||||
|
||||
def test_real_trades_are_weighted_higher(tmp_path):
|
||||
learner = make_learner(tmp_path)
|
||||
learner.learn_from_trade(np.array([1.0, 0.0, 0.0, 0.0]), pnl_quote=12.5)
|
||||
assert learner.stats.trade_samples == 1
|
||||
assert learner.stats.shadow_samples == 0
|
||||
assert learner.buffer.w[0] == pytest.approx(learner.config.trade_sample_weight)
|
||||
|
||||
|
||||
def test_trade_without_features_is_ignored(tmp_path):
|
||||
learner = make_learner(tmp_path)
|
||||
learner.learn_from_trade(None, pnl_quote=1.0)
|
||||
assert learner.stats.samples_seen == 0
|
||||
|
||||
|
||||
def test_wrong_feature_length_is_dropped(tmp_path):
|
||||
learner = make_learner(tmp_path)
|
||||
learner.observe(np.zeros(9), 1.0)
|
||||
assert learner.stats.samples_seen == 0
|
||||
|
||||
|
||||
def test_frozen_learner_scores_but_does_not_train(tmp_path):
|
||||
learner = make_learner(tmp_path)
|
||||
learner.frozen = True
|
||||
before = learner.model.w.copy()
|
||||
for _ in range(50):
|
||||
learner.observe(np.array([1.0, 0.0, 0.0, 0.0]), 1.0)
|
||||
assert np.allclose(learner.model.w, before)
|
||||
assert learner.stats.samples_seen == 50 # Beobachtungen werden trotzdem gesammelt
|
||||
|
||||
|
||||
# --------------------------------------------------- Verzögerte Shadow-Labels
|
||||
|
||||
|
||||
def test_pending_label_resolves_on_target_hit(tmp_path):
|
||||
learner = make_learner(tmp_path)
|
||||
learner.config.label_target_bps = 100.0 # 1 %
|
||||
learner.register_candidate("BTC/USDT", np.ones(4), price=100.0, bar_index=0)
|
||||
assert learner.pending_count == 1
|
||||
|
||||
resolved = learner.resolve_pending("BTC/USDT", 1, high=101.5, low=99.9, close=101.0)
|
||||
assert resolved == 1
|
||||
assert learner.pending_count == 0
|
||||
assert learner.buffer.y[0] == 1.0
|
||||
|
||||
|
||||
def test_pending_label_resolves_on_stop_hit(tmp_path):
|
||||
learner = make_learner(tmp_path)
|
||||
learner.config.label_target_bps = 100.0
|
||||
learner.register_candidate("BTC/USDT", np.ones(4), price=100.0, bar_index=0)
|
||||
learner.resolve_pending("BTC/USDT", 1, high=100.2, low=98.5, close=98.7)
|
||||
assert learner.buffer.y[0] == 0.0
|
||||
|
||||
|
||||
def test_pending_label_expires_after_horizon(tmp_path):
|
||||
learner = make_learner(tmp_path)
|
||||
learner.config.label_horizon_bars = 3
|
||||
learner.config.label_target_bps = 500.0 # wird nicht erreicht
|
||||
learner.register_candidate("BTC/USDT", np.ones(4), price=100.0, bar_index=0)
|
||||
for bar in range(1, 3):
|
||||
learner.resolve_pending("BTC/USDT", bar, 100.1, 99.9, 100.05)
|
||||
assert learner.pending_count == 1
|
||||
learner.resolve_pending("BTC/USDT", 3, 100.1, 99.9, 100.05)
|
||||
assert learner.pending_count == 0
|
||||
assert learner.buffer.y[0] == 1.0 # Schluss über dem Einstieg
|
||||
|
||||
|
||||
def test_pending_labels_are_kept_per_symbol(tmp_path):
|
||||
learner = make_learner(tmp_path)
|
||||
learner.register_candidate("BTC/USDT", np.ones(4), 100.0, 0)
|
||||
learner.register_candidate("ETH/USDT", np.ones(4), 100.0, 0)
|
||||
learner.resolve_pending("BTC/USDT", 1, 200.0, 199.0, 199.5)
|
||||
assert learner.pending_count == 1 # ETH bleibt offen
|
||||
|
||||
|
||||
# ------------------------------------------------------------- Persistenz
|
||||
|
||||
|
||||
def test_save_and_load_round_trip(tmp_path):
|
||||
learner = make_learner(tmp_path)
|
||||
rng = np.random.default_rng(5)
|
||||
for _ in range(200):
|
||||
features = rng.normal(0, 1, 4)
|
||||
learner.observe(features, 1.0 if features[0] > 0 else 0.0)
|
||||
probe = np.array([0.7, -0.2, 0.1, 0.4])
|
||||
expected = learner.score(probe)
|
||||
path = learner.save()
|
||||
assert path.is_file()
|
||||
|
||||
restored = make_learner(tmp_path)
|
||||
assert restored.load() is True
|
||||
assert restored.score(probe) == pytest.approx(expected)
|
||||
assert restored.stats.samples_seen == learner.stats.samples_seen
|
||||
assert len(restored.buffer) == len(learner.buffer)
|
||||
|
||||
|
||||
def test_load_without_file_returns_false(tmp_path):
|
||||
assert make_learner(tmp_path).load() is False
|
||||
|
||||
|
||||
def test_model_with_wrong_feature_count_is_ignored(tmp_path):
|
||||
learner = make_learner(tmp_path)
|
||||
learner.observe(np.zeros(4), 1.0)
|
||||
learner.save()
|
||||
|
||||
other = AdaptiveLearner(
|
||||
LearnerConfig(model_path=str(tmp_path / "model.npz")), n_features=9, seed=1
|
||||
)
|
||||
assert other.load() is False
|
||||
|
||||
|
||||
def test_corrupt_model_file_is_tolerated(tmp_path):
|
||||
path = tmp_path / "model.npz"
|
||||
path.write_bytes(b"kein gueltiges npz")
|
||||
assert make_learner(tmp_path).load() is False
|
||||
|
||||
|
||||
def test_autosave_can_be_disabled(tmp_path):
|
||||
learner = make_learner(tmp_path)
|
||||
learner.autosave = False
|
||||
learner.config.save_every_n_updates = 1
|
||||
for _ in range(50):
|
||||
learner.observe(np.ones(4), 1.0)
|
||||
learner.maybe_save()
|
||||
assert not (tmp_path / "model.npz").exists()
|
||||
|
||||
|
||||
# ------------------------------------------------------------- NullLearner
|
||||
|
||||
|
||||
def test_null_learner_accepts_everything():
|
||||
learner = NullLearner()
|
||||
assert learner.ready is True
|
||||
assert learner.score(np.zeros(3)) == 1.0
|
||||
assert learner.explore() is False
|
||||
learner.observe(np.zeros(3), 1.0)
|
||||
assert learner.snapshot() == {"enabled": False}
|
||||
@@ -0,0 +1,226 @@
|
||||
import pytest
|
||||
|
||||
from trademind.config import RiskConfig
|
||||
from trademind.models import ExitReason, Fill, Position, Side
|
||||
from trademind.portfolio import Portfolio
|
||||
from trademind.risk import RiskManager
|
||||
|
||||
DAY_ONE = 1_700_000_000_000
|
||||
DAY_TWO = DAY_ONE + 86_400_000
|
||||
|
||||
|
||||
def make_fill(symbol="BTC/USDT", side=Side.BUY, amount=0.1, price=30_000.0, fee=3.0, ts=DAY_ONE):
|
||||
return Fill(symbol=symbol, side=side, amount=amount, price=price, fee_quote=fee, timestamp=ts)
|
||||
|
||||
|
||||
def open_position(portfolio: Portfolio, **kwargs) -> Position:
|
||||
return portfolio.open_position(
|
||||
make_fill(**kwargs), stop_loss=None, take_profit=None, features=None,
|
||||
confidence=0.6, exploratory=False,
|
||||
)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ Portfolio
|
||||
|
||||
|
||||
def test_profitable_round_trip_accounts_for_both_fees():
|
||||
portfolio = Portfolio(10_000.0)
|
||||
open_position(portfolio)
|
||||
exit_fill = make_fill(side=Side.SELL, price=31_000.0, fee=3.1)
|
||||
trade = portfolio.close_position(exit_fill, ExitReason.TAKE_PROFIT)
|
||||
|
||||
assert trade.pnl_quote == pytest.approx((31_000 - 30_000) * 0.1 - 6.1)
|
||||
assert trade.is_win
|
||||
assert portfolio.stats.trades == 1
|
||||
assert portfolio.stats.wins == 1
|
||||
assert not portfolio.has_position("BTC/USDT")
|
||||
|
||||
|
||||
def test_losing_trade_is_counted_as_loss():
|
||||
portfolio = Portfolio(10_000.0)
|
||||
open_position(portfolio)
|
||||
trade = portfolio.close_position(make_fill(side=Side.SELL, price=29_000.0), ExitReason.STOP_LOSS)
|
||||
assert trade.pnl_quote < 0
|
||||
assert portfolio.stats.losses == 1
|
||||
assert portfolio.stats.win_rate == 0.0
|
||||
|
||||
|
||||
def test_profit_factor_and_expectancy():
|
||||
portfolio = Portfolio(10_000.0)
|
||||
for exit_price, reason in ((31_000.0, ExitReason.TAKE_PROFIT), (29_500.0, ExitReason.STOP_LOSS)):
|
||||
open_position(portfolio)
|
||||
portfolio.close_position(make_fill(side=Side.SELL, price=exit_price, fee=0.0), reason)
|
||||
stats = portfolio.stats
|
||||
assert stats.gross_profit == pytest.approx(97.0) # 100 − 3 Einstiegsgebühr
|
||||
assert stats.gross_loss == pytest.approx(53.0) # 50 + 3
|
||||
assert stats.profit_factor == pytest.approx(97.0 / 53.0)
|
||||
assert stats.expectancy == pytest.approx((97.0 - 53.0) / 2)
|
||||
|
||||
|
||||
def test_exposure_and_equity_use_mark_prices():
|
||||
portfolio = Portfolio(10_000.0)
|
||||
open_position(portfolio, amount=0.1, price=30_000.0)
|
||||
portfolio.update_mark("BTC/USDT", 32_000.0)
|
||||
assert portfolio.exposure() == pytest.approx(3_200.0)
|
||||
assert portfolio.equity(7_000.0) == pytest.approx(10_200.0)
|
||||
assert portfolio.unrealized_pnl() == pytest.approx(200.0)
|
||||
|
||||
|
||||
def test_drawdown_tracks_the_peak():
|
||||
portfolio = Portfolio(10_000.0)
|
||||
portfolio.record_equity(DAY_ONE, 12_000.0)
|
||||
portfolio.record_equity(DAY_ONE + 1000, 9_000.0)
|
||||
assert portfolio.peak_equity == pytest.approx(12_000.0)
|
||||
assert portfolio.max_drawdown == pytest.approx(0.25)
|
||||
|
||||
|
||||
def test_daily_pnl_resets_on_a_new_utc_day():
|
||||
portfolio = Portfolio(10_000.0)
|
||||
portfolio.record_equity(DAY_ONE, 10_000.0)
|
||||
portfolio.record_equity(DAY_ONE + 3_600_000, 9_500.0)
|
||||
assert portfolio.daily_pnl_pct(9_500.0) == pytest.approx(-0.05)
|
||||
portfolio.record_equity(DAY_TWO, 9_500.0)
|
||||
assert portfolio.daily_pnl_pct(9_500.0) == pytest.approx(0.0)
|
||||
|
||||
|
||||
def test_cooldown_counts_down_per_bar():
|
||||
portfolio = Portfolio(10_000.0)
|
||||
portfolio.start_cooldown("BTC/USDT", 2)
|
||||
assert portfolio.in_cooldown("BTC/USDT")
|
||||
portfolio.on_new_bar("BTC/USDT", 1.0, 1.0, 1.0)
|
||||
assert portfolio.in_cooldown("BTC/USDT")
|
||||
portfolio.on_new_bar("BTC/USDT", 1.0, 1.0, 1.0)
|
||||
assert not portfolio.in_cooldown("BTC/USDT")
|
||||
|
||||
|
||||
def test_bars_held_increases_while_a_position_is_open():
|
||||
portfolio = Portfolio(10_000.0)
|
||||
open_position(portfolio)
|
||||
for _ in range(3):
|
||||
portfolio.on_new_bar("BTC/USDT", 30_500.0, 29_800.0, 30_200.0)
|
||||
assert portfolio.positions["BTC/USDT"].bars_held == 3
|
||||
assert portfolio.positions["BTC/USDT"].highest_price == pytest.approx(30_500.0)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------- Risiko
|
||||
|
||||
|
||||
def test_position_size_respects_the_position_cap():
|
||||
risk = RiskManager(RiskConfig(max_position_pct=0.2, max_total_exposure_pct=1.0))
|
||||
portfolio = Portfolio(10_000.0)
|
||||
amount, reason = risk.position_size(portfolio, 10_000.0, price=100.0)
|
||||
assert reason == ""
|
||||
assert amount == pytest.approx(20.0) # 2000 USDT / 100
|
||||
|
||||
|
||||
def test_position_size_respects_the_exposure_cap():
|
||||
risk = RiskManager(RiskConfig(max_position_pct=0.5, max_total_exposure_pct=0.6))
|
||||
portfolio = Portfolio(10_000.0)
|
||||
open_position(portfolio, amount=0.15, price=30_000.0) # 4500 belegt
|
||||
portfolio.update_mark("BTC/USDT", 30_000.0)
|
||||
amount, _ = risk.position_size(portfolio, 5_500.0, price=100.0)
|
||||
assert amount == pytest.approx(15.0) # 0,6 × 10 000 − 4 500 = 1 500
|
||||
|
||||
|
||||
def test_position_size_rejected_below_minimum_notional():
|
||||
risk = RiskManager(RiskConfig(max_position_pct=0.2, min_notional=100.0))
|
||||
amount, reason = risk.position_size(Portfolio(100.0), 100.0, price=50.0)
|
||||
assert amount == 0.0
|
||||
assert "Minimum" in reason
|
||||
|
||||
|
||||
def test_position_size_rejected_below_exchange_minimum():
|
||||
risk = RiskManager(RiskConfig(max_position_pct=1.0, min_notional=1.0))
|
||||
amount, reason = risk.position_size(Portfolio(50.0), 50.0, price=30_000.0, min_amount=0.01)
|
||||
assert amount == 0.0
|
||||
assert "Börsen-Minimum" in reason
|
||||
|
||||
|
||||
def test_can_open_blocks_duplicates_cooldown_and_limits():
|
||||
risk = RiskManager(RiskConfig(max_open_positions=1))
|
||||
portfolio = Portfolio(10_000.0)
|
||||
assert risk.can_open("BTC/USDT", portfolio, 10_000.0, 30_000.0)
|
||||
|
||||
open_position(portfolio)
|
||||
assert not risk.can_open("BTC/USDT", portfolio, 7_000.0, 30_000.0) # schon offen
|
||||
assert not risk.can_open("ETH/USDT", portfolio, 7_000.0, 2_000.0) # Positionslimit
|
||||
|
||||
portfolio.positions.clear()
|
||||
portfolio.start_cooldown("BTC/USDT", 3)
|
||||
decision = risk.can_open("BTC/USDT", portfolio, 10_000.0, 30_000.0)
|
||||
assert not decision and "Cooldown" in decision.reason
|
||||
|
||||
|
||||
def test_stop_levels_are_derived_from_atr():
|
||||
risk = RiskManager(RiskConfig(stop_loss_atr_mult=2.0, take_profit_atr_mult=3.0))
|
||||
stop, target = risk.stop_levels(100.0, atr=2.0)
|
||||
assert stop == pytest.approx(96.0)
|
||||
assert target == pytest.approx(106.0)
|
||||
|
||||
|
||||
def test_stop_levels_can_be_switched_off():
|
||||
risk = RiskManager(RiskConfig(stop_loss_atr_mult=0.0, take_profit_atr_mult=0.0))
|
||||
assert risk.stop_levels(100.0, 2.0) == (None, None)
|
||||
|
||||
|
||||
def test_exit_prefers_the_stop_when_both_are_touched():
|
||||
risk = RiskManager(RiskConfig())
|
||||
position = Position("BTC/USDT", 1.0, 100.0, DAY_ONE, stop_loss=96.0, take_profit=106.0)
|
||||
reason, price = risk.check_exit(position, high=107.0, low=95.0, close=101.0)
|
||||
assert reason is ExitReason.STOP_LOSS
|
||||
assert price == pytest.approx(96.0)
|
||||
|
||||
|
||||
def test_take_profit_triggers_alone():
|
||||
risk = RiskManager(RiskConfig())
|
||||
position = Position("BTC/USDT", 1.0, 100.0, DAY_ONE, stop_loss=96.0, take_profit=106.0)
|
||||
reason, price = risk.check_exit(position, high=107.0, low=99.0, close=106.5)
|
||||
assert reason is ExitReason.TAKE_PROFIT
|
||||
assert price == pytest.approx(106.0)
|
||||
|
||||
|
||||
def test_trailing_stop_only_moves_upwards():
|
||||
risk = RiskManager(RiskConfig(trailing_stop_atr_mult=1.0))
|
||||
position = Position("BTC/USDT", 1.0, 100.0, DAY_ONE, highest_price=110.0)
|
||||
risk.update_trailing(position, atr=2.0)
|
||||
assert position.trailing_stop == pytest.approx(108.0)
|
||||
position.highest_price = 105.0
|
||||
risk.update_trailing(position, atr=2.0)
|
||||
assert position.trailing_stop == pytest.approx(108.0) # zieht nicht zurück
|
||||
|
||||
|
||||
def test_max_holding_bars_forces_an_exit():
|
||||
risk = RiskManager(RiskConfig(max_holding_bars=5))
|
||||
position = Position("BTC/USDT", 1.0, 100.0, DAY_ONE, bars_held=5)
|
||||
reason, _ = risk.check_exit(position, 101.0, 99.0, 100.0)
|
||||
assert reason is ExitReason.MAX_HOLDING
|
||||
|
||||
|
||||
def test_daily_loss_halts_until_the_next_day():
|
||||
risk = RiskManager(RiskConfig(max_daily_loss_pct=0.05, max_drawdown_pct=0.9))
|
||||
portfolio = Portfolio(10_000.0)
|
||||
portfolio.record_equity(DAY_ONE, 10_000.0)
|
||||
portfolio.record_equity(DAY_ONE + 60_000, 9_400.0)
|
||||
|
||||
assert risk.evaluate_halt(portfolio, 9_400.0, DAY_ONE + 60_000) is not None
|
||||
assert risk.trading_halted
|
||||
assert not risk.force_liquidation()
|
||||
|
||||
portfolio.record_equity(DAY_TWO, 9_400.0)
|
||||
risk.evaluate_halt(portfolio, 9_400.0, DAY_TWO)
|
||||
assert not risk.trading_halted
|
||||
|
||||
|
||||
def test_max_drawdown_halts_permanently_and_liquidates():
|
||||
risk = RiskManager(RiskConfig(max_drawdown_pct=0.2))
|
||||
portfolio = Portfolio(10_000.0)
|
||||
portfolio.record_equity(DAY_ONE, 10_000.0)
|
||||
portfolio.record_equity(DAY_ONE + 1000, 7_000.0)
|
||||
|
||||
reason = risk.evaluate_halt(portfolio, 7_000.0, DAY_ONE + 1000)
|
||||
assert reason and "Drawdown" in reason
|
||||
assert risk.force_liquidation()
|
||||
|
||||
portfolio.record_equity(DAY_TWO, 7_000.0)
|
||||
risk.evaluate_halt(portfolio, 7_000.0, DAY_TWO)
|
||||
assert risk.trading_halted # bleibt bis zum Neustart gesperrt
|
||||
@@ -0,0 +1,218 @@
|
||||
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
|
||||
Reference in New Issue
Block a user