from __future__ import annotations from crypto_spot_bot.learning import TradeLearner from crypto_spot_bot.models import Trade, utc_now from crypto_spot_bot.storage import Storage def test_trade_learner_penalizes_losing_symbol_pattern(make_settings, tmp_path) -> None: settings = make_settings(tmp_path, learning_min_samples=2) storage = Storage(settings.database_path) for value in (-0.4, -0.2): storage.insert_trade( Trade( id=None, symbol="BTCUSDT", side="SELL", qty=1, entry_price=100, exit_price=99, net_pnl=value, reason="test", entry_pattern="пробой вниз", entry_confidence=0.7, opened_at=utc_now(), closed_at=utc_now(), ) ) learner = TradeLearner(settings, storage) learner.refresh() adjustment = learner.adjustment_for("BTCUSDT", "пробой вниз") assert adjustment.sample_size >= 4 assert adjustment.confidence_adjustment < 0 assert "убыточными" in adjustment.reason def test_trade_learner_builds_adaptive_rules_for_losing_ema_exit(make_settings, tmp_path) -> None: settings = make_settings(tmp_path, learning_min_samples=3) storage = Storage(settings.database_path) for value in (-0.05, -0.04, -0.03): storage.insert_trade( Trade( id=None, symbol="BTCUSDT", side="SELL", qty=1, entry_price=100, exit_price=99.9, net_pnl=value, reason="краткосрочный тренд ослаб ниже EMA50", entry_pattern="нейтрально", entry_confidence=0.7, opened_at=utc_now(), closed_at=utc_now(), ) ) learner = TradeLearner(settings, storage) state = learner.refresh() rules = learner.rules_for("BTCUSDT", "нейтрально") assert state.adaptive_rules["risk_mode"] == "selective" assert state.adaptive_rules["trade_permission"] == "selective_growth" assert state.adaptive_rules["reduce_exposure"] is False assert rules["ema_exit_mode"] == "profit_only" assert rules["effective_entry_threshold_adjustment"] > 0 assert rules["min_hold_seconds"] == settings.min_hold_seconds def test_trade_learner_blocks_only_bad_symbol_pattern(make_settings, tmp_path) -> None: settings = make_settings(tmp_path, learning_min_samples=3, starting_balance_usdt=100, max_total_exposure_usdt=80) storage = Storage(settings.database_path) for _ in range(9): storage.insert_trade( Trade( id=None, symbol="ETHUSDT", side="SELL", qty=1, entry_price=100, exit_price=99, net_pnl=-0.10, reason="краткосрочный тренд ослаб ниже EMA50", entry_pattern="нейтрально", entry_confidence=0.7, opened_at=utc_now(), closed_at=utc_now(), ) ) learner = TradeLearner(settings, storage) state = learner.refresh() rules = state.adaptive_rules assert rules["trade_permission"] == "selective_growth" assert rules["reduce_exposure"] is False assert rules["bad_market_entry_block"] is False assert rules["target_total_exposure_usdt"] == settings.max_total_exposure_usdt assert rules["blocked_symbols"] == ["ETHUSDT"] assert rules["blocked_patterns"] == ["нейтрально"] assert rules["validation"]["status"] == "accepted" assert rules["validation"]["avoided_loss_usdt"] > 0 def test_trade_learner_scales_positive_expectancy_setups(make_settings, tmp_path) -> None: settings = make_settings(tmp_path, learning_min_samples=3, learning_max_position_multiplier=1.5) storage = Storage(settings.database_path) for _ in range(4): storage.insert_trade( Trade( id=None, symbol="SOLUSDT", side="SELL", qty=1, entry_price=100, exit_price=101, net_pnl=0.08, reason="test", entry_pattern="пробой вверх", entry_confidence=0.8, opened_at=utc_now(), closed_at=utc_now(), ) ) learner = TradeLearner(settings, storage) state = learner.refresh() rules = learner.rules_for("SOLUSDT", "пробой вверх") assert state.adaptive_rules["trade_permission"] == "positive_expectancy" assert state.adaptive_rules["risk_mode"] == "growth" assert rules["effective_entry_threshold_adjustment"] < 0 assert rules["symbol_position_size_multiplier"] > 1.0 assert rules["pattern_position_size_multiplier"] > 1.0 assert 1.0 < rules["effective_position_size_multiplier"] <= settings.learning_max_position_multiplier