import os import sys from datetime import datetime import pandas as pd import requests PROJECT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) if PROJECT_DIR not in sys.path: sys.path.insert(0, PROJECT_DIR) from app.services import altcoin_screener def test_fetch_all_tickers_filters_stable_and_fiat_suffixes(monkeypatch): monkeypatch.setattr( altcoin_screener.exchange, "fetch_tickers", lambda: { "BTC/USDT": {"last": 1, "percentage": 1, "quoteVolume": 100}, "ETH/USDT": {"last": 2, "percentage": 2, "quoteVolume": 200}, "BNB/USDT": {"last": 3, "percentage": 3, "quoteVolume": 300}, "RLUSD/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "BFUSD/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "EUR/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "AI/USDT": {"last": 1, "percentage": 5, "quoteVolume": 1000}, "USD1/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "U/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "XUSD/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "FRAX/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "LUSD/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "GUSD/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "SUSD/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "USDD/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "EURS/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "AUD/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, }, ) monkeypatch.setattr( altcoin_screener, "_fetch_spot_24h_tickers", lambda: { "BTC/USDT": {"last": 1, "percentage": 1, "quoteVolume": 100}, "ETH/USDT": {"last": 2, "percentage": 2, "quoteVolume": 200}, "BNB/USDT": {"last": 3, "percentage": 3, "quoteVolume": 300}, "RLUSD/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "BFUSD/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "EUR/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "AI/USDT": {"last": 1, "percentage": 5, "quoteVolume": 1000}, "USD1/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "U/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "XUSD/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "FRAX/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "LUSD/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "GUSD/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "SUSD/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "USDD/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "EURS/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, "AUD/USDT": {"last": 1, "percentage": 0.1, "quoteVolume": 100}, }, ) monkeypatch.setattr(altcoin_screener, "get_active_static_exclusions", lambda symbols: {}) monkeypatch.setattr(altcoin_screener, "_fetch_spot_exchange_statuses", lambda: {}) monkeypatch.setattr(altcoin_screener, "record_universe_decisions", lambda items: len(list(items or []))) pairs = altcoin_screener.fetch_all_tickers() assert "AI/USDT" in pairs assert "BTC/USDT" in pairs assert "ETH/USDT" in pairs assert "BNB/USDT" in pairs assert "RLUSD/USDT" not in pairs assert "BFUSD/USDT" not in pairs assert "EUR/USDT" not in pairs assert "USD1/USDT" not in pairs assert "U/USDT" not in pairs assert "XUSD/USDT" not in pairs assert "FRAX/USDT" not in pairs assert "LUSD/USDT" not in pairs assert "GUSD/USDT" not in pairs assert "SUSD/USDT" not in pairs assert "USDD/USDT" not in pairs assert "EURS/USDT" not in pairs assert "AUD/USDT" not in pairs def test_fetch_all_tickers_filters_inactive_and_stale_markets(monkeypatch): fresh_time = datetime.utcnow().isoformat(timespec="seconds") + "Z" monkeypatch.setattr( altcoin_screener, "_fetch_spot_24h_tickers", lambda: { "CREAM/USDT": {"last": 2.1, "percentage": 65, "quoteVolume": 200000, "datetime": "2026-05-08T15:24:40.529"}, "PNT/USDT": {"last": 0.03, "percentage": 45, "quoteVolume": 200000, "datetime": fresh_time}, "FIDA/USDT": {"last": 0.02, "percentage": 35, "quoteVolume": 2000000, "datetime": fresh_time}, "DEAD/USDT": {"last": 1.0, "percentage": 1, "quoteVolume": 2000000, "datetime": fresh_time}, }, ) monkeypatch.setattr( altcoin_screener, "_fetch_spot_exchange_statuses", lambda: {"DEAD/USDT": {"status": "BREAK", "isSpotTradingAllowed": False}}, ) monkeypatch.setattr( altcoin_screener, "get_active_static_exclusions", lambda symbols: {"PNT/USDT": {"reason_code": "invalid_pair", "reason_label": "缓存交易对异常", "reason_type": "static"}}, ) monkeypatch.setattr(altcoin_screener, "record_universe_decisions", lambda items: len(list(items or []))) pairs = altcoin_screener.fetch_all_tickers() assert "CREAM/USDT" not in pairs assert "PNT/USDT" not in pairs assert "DEAD/USDT" not in pairs assert "FIDA/USDT" in pairs exclusions = getattr(altcoin_screener.fetch_all_tickers, "last_universe_exclusions", []) assert any(x["symbol"] == "CREAM/USDT" and x["reason_code"] == "stale_ticker" for x in exclusions) assert any(x["symbol"] == "PNT/USDT" and x.get("cache_hit") for x in exclusions) assert any(x["symbol"] == "DEAD/USDT" and x["reason_code"] == "inactive_market" for x in exclusions) def test_fetch_spot_24h_tickers_uses_cache_when_dns_fails(monkeypatch, tmp_path): monkeypatch.setattr(altcoin_screener, "EXCHANGE_CACHE_DIR", tmp_path) cached_data = [ { "symbol": "AIUSDT", "lastPrice": "1.23", "priceChangePercent": "8.5", "quoteVolume": "1234567", "highPrice": "1.4", "lowPrice": "1.0", "closeTime": 1770000000000, } ] altcoin_screener._write_spot_24h_ticker_cache(cached_data) def fail_get(*args, **kwargs): raise requests.exceptions.ConnectionError("dns failed") monkeypatch.setattr(altcoin_screener.requests, "get", fail_get) tickers = altcoin_screener._fetch_spot_24h_tickers() assert tickers["AI/USDT"]["last"] == 1.23 assert tickers["AI/USDT"]["percentage"] == 8.5 assert tickers["AI/USDT"]["quoteVolume"] == 1234567 def test_dynamic_low_turnover_cache_only_skips_while_volume_still_low(): cached = { "SLOW/USDT": { "reason_type": "dynamic", "reason_code": "low_turnover", "evidence": {"min_volume": 2_000_000}, } } assert altcoin_screener._is_cached_dynamic_exclusion("SLOW/USDT", {"volume_24h": 900_000}, cached) assert not altcoin_screener._is_cached_dynamic_exclusion("SLOW/USDT", {"volume_24h": 3_000_000}, cached) def test_kline_scan_selection_is_rule_based_without_default_count_cap(monkeypatch): monkeypatch.setattr(altcoin_screener, "get_burst_threshold", lambda symbol: 4) monkeypatch.setattr(altcoin_screener, "is_meme_coin", lambda symbol: False) tickers = { "AAA/USDT": {"volume_24h": 20_000_000, "change_24h": 1}, "HOT/USDT": {"volume_24h": 10_000_000, "change_24h": 8}, "OLD/USDT": {"volume_24h": 6_000_000, "change_24h": 0.5}, "LOW/USDT": {"volume_24h": 500_000, "change_24h": 99}, } selected = altcoin_screener._rule_based_kline_scan_symbols( tickers, recently_screened={"OLD/USDT"}, min_volume=5_000_000, emergency_max=0, ) capped = altcoin_screener._rule_based_kline_scan_symbols( tickers, recently_screened={"OLD/USDT"}, min_volume=5_000_000, emergency_max=2, ) assert "LOW/USDT" not in selected assert {"AAA/USDT", "HOT/USDT", "OLD/USDT"} <= set(selected) assert len(selected) == 3 assert len(capped) == 2 def test_discovery_priority_queue_promotes_top_gainers_before_kline_scan(monkeypatch): monkeypatch.setattr(altcoin_screener, "get_burst_threshold", lambda symbol: 4) monkeypatch.setattr(altcoin_screener, "is_meme_coin", lambda symbol: False) cfg = { "main_min_volume_usd": 5_000_000, "discovery_min_volume_usd": 2_000_000, "short_tf_high_volume_usd": 20_000_000, "tier_a_min_score": 45, "tier_b_min_score": 20, } tickers = { "SLEEP/USDT": {"price": 1, "volume_24h": 30_000_000, "change_24h": 0.2, "high_24h": 1.01, "low_24h": 0.99}, "MOVE/USDT": {"price": 1, "volume_24h": 3_000_000, "change_24h": 12, "high_24h": 1.16, "low_24h": 0.96}, "LOW/USDT": {"price": 1, "volume_24h": 300_000, "change_24h": 50, "high_24h": 1.8, "low_24h": 0.9}, } queue = altcoin_screener._build_discovery_priority_queue(tickers, recently_screened=set(), cfg=cfg) by_symbol = {item["symbol"]: item for item in queue} assert by_symbol["MOVE/USDT"]["tier"] == "A" assert by_symbol["SLEEP/USDT"]["tier"] in {"B", "A"} assert "LOW/USDT" not in by_symbol def test_discovery_tier_selection_respects_a_b_budgets(): queue = [ {"symbol": "A1/USDT", "tier": "A", "score": 90}, {"symbol": "A2/USDT", "tier": "A", "score": 80}, {"symbol": "B1/USDT", "tier": "B", "score": 30}, {"symbol": "B2/USDT", "tier": "B", "score": 25}, {"symbol": "C1/USDT", "tier": "C", "score": 5}, ] selected = altcoin_screener._select_discovery_tiers(queue, tier_a_budget=1, tier_b_budget=1) bypass = altcoin_screener._select_discovery_tiers( queue, tier_a_budget=1, tier_b_budget=0, extra_tiers=("B", "C"), extra_budget=2, ) assert selected == ["A1/USDT", "B1/USDT"] assert bypass == ["A1/USDT", "B1/USDT", "B2/USDT"] def _mock_weights(): return { "量价齐飞": 5, "N倍放量": 5, "连续3x放量": 4, "布林收窄": 3, "静K蓄力": 2, "Q≥7供给区突破": 4, "Q7供给区突破": 4, "动K(阳)+量递增": 3, "动K阳量递增": 3, "连续K加速": 3, "板块联动": 3, "大户偏多": 1, "静K→动K转折": 4, "静K动K转折": 4, "1H放量(量价背离)": 1, } def test_volume_price_fly_accepts_two_consecutive_4x_bars(monkeypatch): monkeypatch.setattr( altcoin_screener, "vp_fly_params", lambda: {"vol_ratio_min": 5.0, "body_ratio_min": 0.70, "consecutive_relaxed_vol_ratio_min": 4.0}, ) rows = [] for i in range(20): rows.append({"open": 1.0, "high": 1.03, "low": 0.99, "close": 1.01, "volume": 100.0}) rows.extend([ {"open": 1.00, "high": 1.10, "low": 0.99, "close": 1.09, "volume": 650.0}, {"open": 1.09, "high": 1.20, "low": 1.08, "close": 1.18, "volume": 670.0}, ]) df = pd.DataFrame(rows) vp = altcoin_screener.detect_volume_price_fly(df) assert vp["vp_fly_count"] == 2 assert len(vp["vp_fly_details"]) == 2 def test_short_timeframe_ignition_detects_recent_15m_start(): rows = [] price = 1.0 for _ in range(40): rows.append({"open": price, "high": price * 1.004, "low": price * 0.996, "close": price * 1.001, "volume": 100.0}) price *= 1.001 rows.extend([ {"open": price, "high": price * 1.01, "low": price * 0.998, "close": price * 1.008, "volume": 160.0}, {"open": price * 1.008, "high": price * 1.035, "low": price * 1.004, "close": price * 1.031, "volume": 420.0}, ]) df = pd.DataFrame(rows) result = altcoin_screener.detect_short_timeframe_ignition( df, "15m", {"recent_bars": 8, "max_trigger_age_bars": 2, "min_vol_ratio": 2.5, "min_body_ratio": 0.45, "min_gain_pct": 0.8}, ) assert result["found"] is True assert result["timeframe"] == "15m" assert result["trigger"]["age_bars"] <= 2 def test_layer1_keeps_high_momentum_breakout_without_5x_vp(monkeypatch): monkeypatch.setattr(altcoin_screener, "fetch_all_tickers", lambda: { "CREAM/USDT": {"price": 2.1, "change_24h": 12.0, "volume_24h": 8000000}, }) monkeypatch.setattr(altcoin_screener, "fetch_funding_rates", lambda: {}) monkeypatch.setattr(altcoin_screener, "is_meme_coin", lambda symbol: False) monkeypatch.setattr(altcoin_screener, "get_burst_threshold", lambda symbol: 20) monkeypatch.setattr(altcoin_screener, "funding_rate_params", lambda: {"long_extreme": 0.001, "short_extreme": -0.0005}) monkeypatch.setattr(altcoin_screener, "detect_bollinger_squeeze", lambda df: None) monkeypatch.setattr(altcoin_screener, "vp_fly_params", lambda: {"vol_ratio_min": 5.0, "body_ratio_min": 0.70, "consecutive_relaxed_vol_ratio_min": 4.0}) monkeypatch.setattr(altcoin_screener, "get_dynamic_weights", lambda: { "量价齐飞": 5, "N倍放量(≥10x)": 6, "连续3x放量(≥3根)": 4, "布林收窄": 3, "静K蓄力": 2, "1H放量(量价背离)": 1, }) h1_rows = [] for _ in range(20): h1_rows.append([0, 1.0, 1.03, 0.99, 1.01, 100.0]) h1_rows.extend([ [0, 1.00, 1.10, 0.99, 1.09, 650.0], [0, 1.09, 1.20, 1.08, 1.18, 670.0], ]) h4_rows = [] price = 1.0 for _ in range(30): h4_rows.append([0, price, price * 1.01, price * 0.99, price * 1.002, 100.0]) price *= 1.001 def fake_fetch_klines(symbol, timeframe, limit=100): data = h1_rows if timeframe == '1h' else h4_rows return pd.DataFrame(data, columns=['timestamp','open','high','low','close','volume']) monkeypatch.setattr(altcoin_screener, "fetch_klines", fake_fetch_klines) candidates = altcoin_screener.layer1_coarse_filter() vp = altcoin_screener.detect_volume_price_fly(fake_fetch_klines('CREAM/USDT', '1h')) assert vp["vp_fly_count"] == 2 assert "CREAM/USDT" in candidates assert any("量价齐飞" in s for s in candidates["CREAM/USDT"]["anomalies"]) assert any("连续2根量价齐飞K" in s for s in candidates["CREAM/USDT"]["anomalies"]) def test_static_accumulation_bypass_promotes_expired_to_accumulate(monkeypatch): monkeypatch.setattr(altcoin_screener, "get_dynamic_weights", _mock_weights) monkeypatch.setattr(altcoin_screener, "state_score_thresholds", lambda: (8, 10, 3)) monkeypatch.setattr( altcoin_screener, "get_screener_section", lambda name=None: { "sector_rotation": { "bonus_weight": 0, "min_non_sector_signals_for_accelerate": 2, "sector_only_max_state": "蓄力", }, "static_accumulation_bypass": { "min_score": 2, "min_vol_ratio": 1.0, "min_static_count": 3, }, }.get(name, {}), ) monkeypatch.setattr(altcoin_screener, "fetch_top_trader_ratio", lambda symbol: None) monkeypatch.setattr(altcoin_screener, "log_screening", lambda **kwargs: None) monkeypatch.setattr(altcoin_screener, "get_sector_for_coin", lambda symbol: []) monkeypatch.setattr(altcoin_screener, "dynamic_leader_detection", lambda perf: {}) monkeypatch.setattr(altcoin_screener, "SECTOR_MEMBERS", {}) qualified, _, _ = altcoin_screener.layer2_fine_filter({ "PNT/USDT": { "anomaly_score": 2, "price": 1.0, "change_24h": 4.0, "funding_rate": 0.0, "is_meme": False, "vp_data": None, "bb_data": None, "static_accumulation": {"static_count": 5, "vol_ratio": 1.1}, "h4_df": None, } }) assert qualified["PNT/USDT"]["state"] == "蓄力" assert qualified["PNT/USDT"]["base_state"] == "过期" assert qualified["PNT/USDT"]["force_reason"] == "静K蓄力旁路" assert any("静K蓄力旁路入池" in s for s in qualified["PNT/USDT"]["signals"]) def test_static_accumulation_requires_resonance_when_configured(monkeypatch): monkeypatch.setattr(altcoin_screener, "get_dynamic_weights", _mock_weights) monkeypatch.setattr(altcoin_screener, "state_score_thresholds", lambda: (8, 10, 3)) logged = [] monkeypatch.setattr( altcoin_screener, "get_screener_section", lambda name=None: { "sector_rotation": { "bonus_weight": 0, "min_non_sector_signals_for_accelerate": 2, "sector_only_max_state": "蓄力", }, "static_accumulation_bypass": { "min_score": 2, "min_vol_ratio": 1.0, "min_static_count": 3, "require_resonance": True, "min_resonance_signals": 2, }, }.get(name, {}), ) monkeypatch.setattr(altcoin_screener, "fetch_top_trader_ratio", lambda symbol: None) monkeypatch.setattr(altcoin_screener, "log_screening", lambda **kwargs: logged.append(kwargs)) monkeypatch.setattr(altcoin_screener, "get_sector_for_coin", lambda symbol: []) monkeypatch.setattr(altcoin_screener, "dynamic_leader_detection", lambda perf: {}) monkeypatch.setattr(altcoin_screener, "SECTOR_MEMBERS", {}) qualified, _, _ = altcoin_screener.layer2_fine_filter({ "QUIET/USDT": { "anomaly_score": 2, "price": 1.0, "change_24h": 1.0, "funding_rate": 0.0, "is_meme": False, "vp_data": None, "bb_data": None, "static_accumulation": {"static_count": 5, "vol_ratio": 1.1}, "h4_df": None, } }) assert "QUIET/USDT" not in qualified reject = next(item for item in logged if item["symbol"] == "QUIET/USDT") assert reject["state"] == "过期" assert reject["detail"]["static_bypass_resonance"] == [] def test_strong_static_accumulation_can_promote_to_accelerate(monkeypatch): monkeypatch.setattr(altcoin_screener, "get_dynamic_weights", _mock_weights) monkeypatch.setattr(altcoin_screener, "state_score_thresholds", lambda: (8, 10, 3)) monkeypatch.setattr( altcoin_screener, "get_screener_section", lambda name=None: { "sector_rotation": { "bonus_weight": 0, "min_non_sector_signals_for_accelerate": 2, "sector_only_max_state": "蓄力", }, "static_accumulation_bypass": { "min_score": 2, "min_vol_ratio": 1.0, "min_static_count": 3, "direct_accelerate": { "enabled": True, "min_static_count": 10, "min_vol_ratio": 1.25, "min_score": 5, }, }, }.get(name, {}), ) monkeypatch.setattr(altcoin_screener, "fetch_top_trader_ratio", lambda symbol: None) monkeypatch.setattr(altcoin_screener, "log_screening", lambda **kwargs: None) monkeypatch.setattr(altcoin_screener, "get_sector_for_coin", lambda symbol: []) monkeypatch.setattr(altcoin_screener, "dynamic_leader_detection", lambda perf: {}) monkeypatch.setattr(altcoin_screener, "SECTOR_MEMBERS", {}) qualified, _, _ = altcoin_screener.layer2_fine_filter({ "PNT/USDT": { "anomaly_score": 5, "price": 1.0, "change_24h": 4.0, "funding_rate": 0.0, "is_meme": False, "vp_data": None, "bb_data": None, "static_accumulation": {"static_count": 12, "vol_ratio": 1.4}, "h4_df": None, } }) assert qualified["PNT/USDT"]["state"] == "加速" assert qualified["PNT/USDT"]["base_state"] == "蓄力" assert qualified["PNT/USDT"]["force_reason"] == "强静K蓄力直升加速" assert any("强静K蓄力直升加速" in s for s in qualified["PNT/USDT"]["signals"]) assert qualified["PNT/USDT"]["candidate_stage"] == "qualified_candidate" assert qualified["PNT/USDT"]["next_stage"] == "trade_confirm" assert "rec_id" not in qualified["PNT/USDT"] def test_layer1_logs_coarse_candidate_details(monkeypatch): logged = [] h4_df = pd.DataFrame({ "open": [1.0] * 24, "high": [1.01] * 24, "low": [0.99] * 24, "close": [1.0] * 24, "volume": [1000] * 24, }) monkeypatch.setattr(altcoin_screener, "fetch_all_tickers", lambda: { "DOGE/USDT": {"volume_24h": 20_000_000, "change_24h": 3.5, "price": 0.1}, }) monkeypatch.setattr(altcoin_screener, "fetch_funding_rates", lambda: {}) monkeypatch.setattr(altcoin_screener, "get_dynamic_weights", _mock_weights) monkeypatch.setattr(altcoin_screener, "is_meme_coin", lambda symbol: False) monkeypatch.setattr(altcoin_screener, "get_burst_threshold", lambda symbol: 20) monkeypatch.setattr(altcoin_screener, "fetch_klines", lambda symbol, timeframe, limit=200: h4_df if timeframe == "4h" else None) monkeypatch.setattr( altcoin_screener, "detect_static_accumulation", lambda symbol, df: {"static_count": 6, "vol_ratio": 1.5}, ) monkeypatch.setattr( altcoin_screener, "get_screener_section", lambda name=None: { "static_accumulation_bypass": {"min_volume_24h": 1_000_000, "min_vol_ratio": 1.2}, "higher_lows": {"enabled": False}, "compression_surge": {"enabled": False}, "sentiment": {"enabled": False}, }.get(name, {}), ) monkeypatch.setattr(altcoin_screener.exchange, "fapiPublicGetTicker24hr", lambda: []) monkeypatch.setattr(altcoin_screener, "log_screening", lambda **kwargs: logged.append(kwargs)) candidates = altcoin_screener.layer1_coarse_filter() assert "DOGE/USDT" in candidates assert any(item["layer"] == "粗筛" and item["symbol"] == "DOGE/USDT" for item in logged) def test_layer1_keeps_unseen_top_gainer_as_discovery_candidate(monkeypatch): logged = [] monkeypatch.setattr(altcoin_screener, "fetch_all_tickers", lambda: { "RONIN/USDT": {"volume_24h": 11_000_000, "change_24h": 43.0, "price": 1.2}, }) monkeypatch.setattr(altcoin_screener, "fetch_funding_rates", lambda: {}) monkeypatch.setattr(altcoin_screener, "get_dynamic_weights", _mock_weights) monkeypatch.setattr(altcoin_screener, "is_meme_coin", lambda symbol: False) monkeypatch.setattr(altcoin_screener, "get_burst_threshold", lambda symbol: 5) monkeypatch.setattr(altcoin_screener, "fetch_klines", lambda symbol, timeframe, limit=200: None) monkeypatch.setattr( altcoin_screener, "get_screener_section", lambda name=None: { "static_accumulation_bypass": {"min_volume_24h": 2_000_000, "min_vol_ratio": 1.2}, "higher_lows": {"enabled": False}, "compression_surge": {"enabled": False}, "sentiment": {"enabled": False}, }.get(name, {}), ) monkeypatch.setattr(altcoin_screener.exchange, "fapiPublicGetTicker24hr", lambda: []) monkeypatch.setattr(altcoin_screener, "log_screening", lambda **kwargs: logged.append(kwargs)) candidates = altcoin_screener.layer1_coarse_filter() assert "RONIN/USDT" in candidates cand = candidates["RONIN/USDT"] assert cand["top_gainer_24h"] is True assert cand["top_gainer_chase_risk"] is True assert any("24h强势榜异动" in s for s in cand["anomalies"]) coarse = next(item for item in logged if item["layer"] == "粗筛" and item["symbol"] == "RONIN/USDT") assert coarse["detail"]["candidate_stage"] == "discovery_candidate" assert "cex_top_gainer" in coarse["detail"]["source_types"] assert "cex_top_gainer_24h" in coarse["detail"]["signal_codes"] def test_top_gainer_quality_rejected_as_chase_not_silent(monkeypatch): logged = [] monkeypatch.setattr(altcoin_screener, "get_dynamic_weights", _mock_weights) monkeypatch.setattr(altcoin_screener, "state_score_thresholds", lambda: (6, 9, 3)) monkeypatch.setattr( altcoin_screener, "get_screener_section", lambda name=None: { "static_accumulation_bypass": { "min_score": 2, "min_vol_ratio": 1.2, "min_static_count": 3, }, }.get(name, {}), ) monkeypatch.setattr(altcoin_screener, "fetch_top_trader_ratio", lambda symbol: None) monkeypatch.setattr(altcoin_screener, "log_screening", lambda **kwargs: logged.append(kwargs)) monkeypatch.setattr(altcoin_screener, "get_sector_for_coin", lambda symbol: []) monkeypatch.setattr(altcoin_screener, "dynamic_leader_detection", lambda perf: {}) monkeypatch.setattr(altcoin_screener, "SECTOR_MEMBERS", {}) qualified, _, _ = altcoin_screener.layer2_fine_filter({ "RONIN/USDT": { "anomaly_score": 2, "price": 1.2, "change_24h": 43.0, "volume_24h": 11_000_000, "funding_rate": 0.0, "is_meme": False, "vp_data": None, "bb_data": None, "static_accumulation": None, "h4_df": None, "top_gainer_24h": True, "top_gainer_chase_risk": True, "anomalies": ["24h强势榜异动(43.0%,成交额11.0M)，追高风险待确认"], } }) assert "RONIN/USDT" not in qualified reject = next(item for item in logged if item["layer"] == "细筛" and item["symbol"] == "RONIN/USDT") assert reject["detail"]["candidate_stage"] == "rejected_candidate" assert "high_chase_risk" in reject["detail"]["reject_reason_codes"] assert "low_score" in reject["detail"]["reject_reason_codes"] def test_top_gainer_without_chase_risk_is_not_rejected_only_for_low_score(monkeypatch): logged = [] monkeypatch.setattr(altcoin_screener, "get_dynamic_weights", _mock_weights) monkeypatch.setattr(altcoin_screener, "state_score_thresholds", lambda: (6, 9, 3)) monkeypatch.setattr( altcoin_screener, "get_screener_section", lambda name=None: { "static_accumulation_bypass": { "min_score": 2, "min_vol_ratio": 1.2, "min_static_count": 3, }, }.get(name, {}), ) monkeypatch.setattr(altcoin_screener, "fetch_top_trader_ratio", lambda symbol: None) monkeypatch.setattr(altcoin_screener, "log_screening", lambda **kwargs: logged.append(kwargs)) monkeypatch.setattr(altcoin_screener, "get_sector_for_coin", lambda symbol: []) monkeypatch.setattr(altcoin_screener, "dynamic_leader_detection", lambda perf: {}) monkeypatch.setattr(altcoin_screener, "SECTOR_MEMBERS", {}) qualified, _, _ = altcoin_screener.layer2_fine_filter({ "RONIN/USDT": { "anomaly_score": 2, "price": 1.2, "change_24h": 43.0, "volume_24h": 11_000_000, "funding_rate": 0.0, "is_meme": False, "vp_data": None, "bb_data": None, "static_accumulation": None, "h4_df": None, "top_gainer_24h": True, "top_gainer_chase_risk": False, "anomalies": ["24h强势榜异动(43.0%,成交额11.0M)"], } }) # 低分强势榜候选仍可被保留为异动候选，而不是直接纯拒绝掉。 assert "RONIN/USDT" in qualified fine = next(item for item in logged if item["layer"] == "细筛" and item["symbol"] == "RONIN/USDT") assert fine["detail"]["candidate_stage"] == "qualified_candidate" assert "high_chase_risk" not in fine["detail"]["reject_reason_codes"] if "reject_reason_codes" in fine["detail"] else True