128 lines
4.3 KiB
Python
128 lines
4.3 KiB
Python
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
|