From ec96f9c012ea7d10ebfdaa9e1a9d18e99d259ba6 Mon Sep 17 00:00:00 2001 From: aaron <> Date: Sun, 31 May 2026 23:13:08 +0800 Subject: [PATCH] 1 --- app/services/altcoin_confirm.py | 7 +++++-- app/services/altcoin_screener.py | 6 +++--- app/strategies/short_breakdown.py | 1 + tests/test_multi_strategy_infra.py | 23 ++++++++++++++++++++++- 4 files changed, 31 insertions(+), 6 deletions(-) diff --git a/app/services/altcoin_confirm.py b/app/services/altcoin_confirm.py index bcc713d..ba45dda 100644 --- a/app/services/altcoin_confirm.py +++ b/app/services/altcoin_confirm.py @@ -1727,12 +1727,15 @@ def confirm_burst(symbol, cand): short_entry_plan = None if trade_side == "short": entry_action = "即刻买入" if short_1h.get("retest_rejected") else "等回踩" - retest_zone = short_1h.get("retest_zone") or [] + retest_zone = short_1h.get("retest_zone") or short_1h.get("breakdown_level") or 0 if entry_action == "即刻买入": entry_price = round(float(price), 6) entry_method = "空头1H破位反抽失败，当前可开空" else: - zone_mid = sum(float(x or 0) for x in retest_zone[:2]) / 2 if len(retest_zone) >= 2 else float(short_1h.get("breakdown_level") or price) + if isinstance(retest_zone, (list, tuple)) and len(retest_zone) >= 2: + zone_mid = sum(float(x or 0) for x in retest_zone[:2]) / 2 + else: + zone_mid = float(retest_zone or price) entry_price = round(max(float(price), zone_mid), 6) entry_method = f"等反抽到${entry_price:.4f}(箱体下沿反压)" stop_loss = round(float(short_1h.get("stop_level") or price * 1.055), 6) diff --git a/app/services/altcoin_screener.py b/app/services/altcoin_screener.py index 3634a59..5e2b94e 100644 --- a/app/services/altcoin_screener.py +++ b/app/services/altcoin_screener.py @@ -1217,10 +1217,10 @@ def layer1_coarse_filter(): short_breakdown_1h = detect_breakdown_retest_short_1h(h1_df, change_24h=change) if short_breakdown_1h.get("detected"): - quality = float(short_breakdown_1h.get("quality") or 0) - if quality >= 0.48: + quality_score = float(short_breakdown_1h.get("quality_score") or short_breakdown_1h.get("score") or 0) + if quality_score >= 5: anomalies.extend(short_breakdown_1h.get("signals") or ["1H破位反抽做空结构"]) - anomaly_score += int(short_breakdown_1h.get("score") or 4) + anomaly_score += int(quality_score or 4) # 布林收窄检测（4H级别） if h4_df is not None and len(h4_df) >= 20: diff --git a/app/strategies/short_breakdown.py b/app/strategies/short_breakdown.py index bc93e2a..dec6649 100644 --- a/app/strategies/short_breakdown.py +++ b/app/strategies/short_breakdown.py @@ -72,6 +72,7 @@ def detect_breakdown_retest_short_1h(df, *, change_24h: float = 0.0) -> dict: "stop_level": round(stop_level, 8), "target_1": round(target_1, 8), "quality": quality, + "quality_score": quality_score, "score": quality_score, "retest_rejected": latest_rejected, "relative_weakness": bool(below_ema or float(change_24h or 0) <= -3), diff --git a/tests/test_multi_strategy_infra.py b/tests/test_multi_strategy_infra.py index 7acf3de..924989b 100644 --- a/tests/test_multi_strategy_infra.py +++ b/tests/test_multi_strategy_infra.py @@ -20,7 +20,7 @@ from app.strategies.altcoin_breakout import ( build_volume_ignition_1h_signal, ) from app.strategies.box_retest_4h import build_box_retest_1h_signal -from app.strategies.short_breakdown import build_breakdown_retest_short_1h_signal +from app.strategies.short_breakdown import build_breakdown_retest_short_1h_signal, detect_breakdown_retest_short_1h def test_factor_roles_never_promote_unknown_to_trigger(): @@ -136,6 +136,27 @@ def test_breakdown_retest_short_strategy_is_independent_short_signal(): assert signal["factor_roles"]["breakdown_retest_1h_short"] == "trigger" +def test_short_breakdown_detector_exposes_numeric_quality_score(): + import pandas as pd + + rows = [] + price = 1.0 + for i in range(64): + rows.append({"timestamp": i, "open": price, "high": 1.02, "low": 0.98, "close": price, "volume": 100}) + rows.extend([ + {"timestamp": 64, "open": 1.0, "high": 1.01, "low": 0.95, "close": 0.97, "volume": 180}, + {"timestamp": 65, "open": 0.97, "high": 1.01, "low": 0.96, "close": 0.99, "volume": 140}, + {"timestamp": 66, "open": 0.99, "high": 1.015, "low": 0.955, "close": 0.965, "volume": 180}, + {"timestamp": 67, "open": 0.965, "high": 0.98, "low": 0.94, "close": 0.955, "volume": 200}, + ]) + result = detect_breakdown_retest_short_1h(pd.DataFrame(rows), change_24h=-4) + + assert result["detected"] is True + assert result["quality"] in {"良好", "优质"} + assert isinstance(result["quality_score"], int) + assert result["quality_score"] >= 5 + + def test_strategy_evaluation_recommends_promote_or_pause(): strong = evaluate_strategy_decision({ "signal_count": 24,