import importlib.util import sys import types from pathlib import Path def load_class(module_rel_path: str, class_name: str): base = Path(__file__).resolve().parents[1] / "app" / "crypto_agent" target = base / module_rel_path if "app" not in sys.modules: app_pkg = types.ModuleType("app") app_pkg.__path__ = [str(base.parents[1] / "app")] sys.modules["app"] = app_pkg if "app.crypto_agent" not in sys.modules: crypto_pkg = types.ModuleType("app.crypto_agent") crypto_pkg.__path__ = [str(base)] sys.modules["app.crypto_agent"] = crypto_pkg module_name = f"app.crypto_agent.{target.stem}_test" spec = importlib.util.spec_from_file_location(module_name, target) module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module spec.loader.exec_module(module) return getattr(module, class_name) def test_regime_engine_blocks_middle_of_range_in_ranging_market(): RegimeEngine = load_class("regime_engine.py", "RegimeEngine") engine = RegimeEngine() profile = engine.classify( range_metrics={"regime": "ranging", "bb_squeeze": False}, market_location={"location_tag": "middle_of_range", "relative_to_range": "middle_of_range"}, trend_direction="neutral", trend_strength="weak", derivatives_state={"crowding_regime": "low", "crowding_bias": "neutral"}, reversal_detection={}, trend_stage={}, ) assert profile["regime_key"] == "range" assert profile["tradability"] == "selective" assert "range_reversal" in profile["allowed_setups"] assert any("区间边界" in reason or "位置不佳" in reason for reason in profile["no_trade_reasons"]) def test_setup_policy_allows_only_short_term_range_reversal_in_range_market(): SetupPolicy = load_class("setup_policy.py", "SetupPolicy") policy = SetupPolicy() profile = { "tradability": "selective", "allowed_lanes": ["short_term"], "allowed_setups": ["range_reversal"], } signals = [ { "timeframe": "medium_term", "type": "medium_term", "action": "sell", "entry_type": "limit", "regime": "ranging", "market_location": {"location_tag": "near_range_resistance"}, }, { "timeframe": "short_term", "type": "short_term", "action": "sell", "entry_type": "limit", "regime": "ranging", "market_location": {"location_tag": "near_range_resistance"}, }, ] filtered, reasons = policy.filter_signals(signals, profile) assert len(filtered) == 1 assert filtered[0]["timeframe"] == "short_term" assert filtered[0]["setup_type"] == "range_reversal" assert reasons def test_setup_policy_uses_volume_price_context_for_breakout_and_pullback_setups(): SetupPolicy = load_class("setup_policy.py", "SetupPolicy") policy = SetupPolicy() profile = { "tradability": "selective", "allowed_lanes": ["short_term", "medium_term"], "allowed_setups": ["breakout_confirmation", "trend_continuation_pullback"], } signals = [ { "timeframe": "short_term", "type": "short_term", "action": "buy", "entry_type": "market", "regime": "transitional", "market_location": {"location_tag": "near_long_zone"}, "volume_price_context": { "breakout_quality": "acceptance_breakout_up", "volume_price_state": "bullish_acceptance", }, }, { "timeframe": "medium_term", "type": "medium_term", "action": "buy", "entry_type": "limit", "regime": "weak_trend", "market_location": {"location_tag": "near_long_zone"}, "volume_price_context": { "pullback_quality": "healthy_pullback", }, }, ] filtered, reasons = policy.filter_signals(signals, profile) assert len(filtered) == 2 assert filtered[0]["setup_type"] == "breakout_confirmation" assert filtered[1]["setup_type"] == "trend_continuation_pullback" assert filtered[0]["entry_basis"] == "breakout_acceptance_follow_through" assert "setup=breakout_confirmation" in filtered[0]["setup_basis"] assert not reasons