from app.core.factor_roles import RISK, TRIGGER, factor_role, factor_roles_for_codes from app.core.strategy_contract import StrategySignal, default_main_composite_signal from app.core.strategy_registry import ( BOX_RETEST_1H_STRATEGY, BOX_RETEST_4H_STRATEGY, COMPRESSION_BREAKOUT_4H_STRATEGY, BREAKDOWN_RETEST_SHORT_1H_STRATEGY, INTRADAY_MOMENTUM_15M_STRATEGY, MAIN_COMPOSITE_STRATEGY, VOLUME_IGNITION_1H_STRATEGY, strategy_label, ) from app.db.recommendation_commands import create_recommendation from app.db.paper_trading import _open_trade, _order_payload_from_rec from app.db.strategy_signal_queries import insert_strategy_signal from app.db.strategy_insights import evaluate_strategy_decision from app.strategies.altcoin_breakout import ( build_compression_breakout_4h_signal, build_intraday_momentum_15m_signal, 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, detect_breakdown_retest_short_1h def test_factor_roles_never_promote_unknown_to_trigger(): assert factor_role("box_breakout_pullback_4h") == TRIGGER assert factor_role("box_breakout_pullback_1h") == TRIGGER assert factor_role("false_breakout") == RISK assert factor_role("new_unknown_factor") == "unknown" assert factor_roles_for_codes(["box_breakout_pullback_4h", "new_unknown_factor"]) == { "box_breakout_pullback_4h": "trigger", "new_unknown_factor": "unknown", } def test_default_main_composite_strategy_signal_is_stable(): signal = default_main_composite_signal( symbol="AAA/USDT", score=70, signal_codes=["vp_fly_1h_current"], entry_plan={"entry_action": "观察"}, ).to_json_dict() assert signal["strategy_code"] == MAIN_COMPOSITE_STRATEGY assert signal["strategy_name"] == "综合确认策略" assert signal["factor_roles"]["vp_fly_1h_current"] == "trigger" assert strategy_label(BOX_RETEST_1H_STRATEGY) == "1H箱体突破回踩" assert strategy_label(VOLUME_IGNITION_1H_STRATEGY) == "1H放量突破启动" assert strategy_label(COMPRESSION_BREAKOUT_4H_STRATEGY) == "4H压缩蓄力突破" assert strategy_label(INTRADAY_MOMENTUM_15M_STRATEGY) == "15m日内动量延续" assert strategy_label(BREAKDOWN_RETEST_SHORT_1H_STRATEGY) == "1H破位反抽做空" def test_volume_ignition_strategy_builds_independent_signal(): signal = build_volume_ignition_1h_signal( symbol="VOL/USDT", result={ "score": 8, "signals": ["1H量价齐飞 · 连续放量", "15min即刻入场"], "trigger_context": {"current_triggers": ["15m突破"], "trigger_status": "current"}, "entry_plan": {"entry_action": "可即刻买入"}, }, entry_plan={"entry_action": "可即刻买入", "entry_price": 1.0}, ) payload = signal.to_json_dict() assert payload["strategy_code"] == VOLUME_IGNITION_1H_STRATEGY assert payload["status"] == "candidate" assert payload["trigger"]["factor_code"] == "vp_fly_1h_current" assert payload["factor_roles"]["vp_fly_1h_current"] == "trigger" def test_compression_breakout_strategy_requires_structure_and_breakout_context(): signal = build_compression_breakout_4h_signal( symbol="QUIET/USDT", result={"score": 6, "signals": ["4H静K压缩，突破箱体上沿"], "entry_plan": {"entry_action": "等回踩"}}, entry_plan={"entry_action": "等回踩", "entry_price": 1.0}, ) payload = signal.to_json_dict() assert payload["strategy_code"] == COMPRESSION_BREAKOUT_4H_STRATEGY assert payload["status"] == "candidate" assert payload["factor_roles"]["compression_surge_4h"] == "trigger" assert build_compression_breakout_4h_signal( symbol="NOBOX/USDT", result={"score": 6, "signals": ["1H量价齐飞"], "entry_plan": {"entry_action": "可即刻买入"}}, entry_plan={"entry_action": "可即刻买入"}, ) is None def test_intraday_momentum_strategy_requires_current_trigger(): stale = build_intraday_momentum_15m_signal( symbol="FAST/USDT", result={"score": 7, "signals": ["15m短周期启动"], "entry_plan": {"entry_action": "可即刻买入"}}, entry_plan={"entry_action": "可即刻买入"}, ).to_json_dict() fresh = build_intraday_momentum_15m_signal( symbol="FAST/USDT", result={ "score": 7, "signals": ["15min强突破"], "trigger_context": {"current_triggers": ["15m突破"], "trigger_status": "current"}, "entry_plan": {"entry_action": "可即刻买入"}, }, entry_plan={"entry_action": "可即刻买入"}, ).to_json_dict() assert stale["status"] == "observe" assert "缺少当前低周期触发" in stale["risk_plan"]["risk_reasons"] assert fresh["status"] == "candidate" def test_breakdown_retest_short_strategy_is_independent_short_signal(): signal = build_breakdown_retest_short_1h_signal( symbol="WEAK/USDT", current_price=0.98, detection={ "detected": True, "score": 8, "breakdown_level": 1.0, "retest_zone": 1.0, "stop_level": 1.04, "target_1": 0.9, "quality": "优质", "retest_rejected": True, "relative_weakness": True, }, market_regime={"risk_level": "high", "regime": "risk_off"}, ).to_json_dict() assert signal["strategy_code"] == BREAKDOWN_RETEST_SHORT_1H_STRATEGY assert signal["direction"] == "short" assert signal["entry_plan"]["side"] == "short" assert signal["status"] == "candidate" 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, "opportunity_count": 16, "trade_count": 8, "closed_trade_count": 8, "win_rate_pct": 62.5, "avg_realized_pnl_pct": 3.2, "realized_pnl_usdt": 180, "worst_pnl_pct": -3.5, "order_fill_rate_pct": 45, "trade_conversion_pct": 50, }) weak = evaluate_strategy_decision({ "signal_count": 24, "opportunity_count": 16, "trade_count": 8, "closed_trade_count": 8, "win_rate_pct": 25, "avg_realized_pnl_pct": -2.5, "realized_pnl_usdt": -120, "worst_pnl_pct": -9, "order_fill_rate_pct": 20, "trade_conversion_pct": 50, }) unfilled = evaluate_strategy_decision({ "signal_count": 18, "opportunity_count": 12, "trade_count": 0, "closed_trade_count": 0, }) assert strong["decision"] == "promote" assert strong["evaluation_score"] > weak["evaluation_score"] assert weak["decision"] == "pause" assert unfilled["decision"] == "review_entry_gate" def test_strategy_signal_insert_and_recommendation_lineage(pg_conn): signal = insert_strategy_signal( StrategySignal( strategy_code=BOX_RETEST_4H_STRATEGY, symbol="BOX/USDT", score=10, confidence=80, trigger={"factor_code": "box_breakout_pullback_4h"}, factor_roles={"box_breakout_pullback_4h": "trigger"}, entry_plan={"entry_action": "等回踩", "entry_price": 1.0}, ) ) rec_id = create_recommendation( symbol="BOX/USDT", rec_state="爆发", rec_score=30, entry_price=1.0, stop_loss=0.94, tp1=1.12, signals=["4H箱体突破回踩(箱体上沿 $1, 量2x)"], entry_plan={"entry_action": "等回踩", "entry_price": 1.0, "stop_loss": 0.94, "tp1": 1.12}, strategy_code=signal["strategy_code"], strategy_signal_id=signal["strategy_signal_id"], strategy_snapshot=signal, factor_roles=signal["factor_roles"], ) row = pg_conn.execute("SELECT * FROM recommendation WHERE id=%s", (rec_id,)).fetchone() assert row["strategy_code"] == BOX_RETEST_4H_STRATEGY assert row["strategy_signal_id"] == signal["strategy_signal_id"] assert "box_breakout_pullback_4h" in row["factor_roles_json"] def test_box_retest_strategy_preserves_zero_age_as_fresh(): signal = build_box_retest_1h_signal( symbol="FRESH/USDT", current_price=1.01, detection={ "detected": True, "score": 10, "entry_zone": 1.0, "stop_level": 0.94, "quality": "优质", "pullback_age_bars": 0, }, market_regime={"regime": "altcoin_rotation", "risk_level": "medium"}, ) payload = signal.to_json_dict() assert payload["status"] == "candidate" assert payload["trigger"]["pullback_age_bars"] == 0 assert payload["risk_plan"]["risk_reasons"] == [] def test_paper_order_and_trade_inherit_strategy_lineage(pg_conn): rec = { "id": 1, "symbol": "BOX/USDT", "rec_score": 100, "entry_price": 1.0, "stop_loss": 0.94, "tp1": 1.12, "tp2": 1.2, "execution_status": "buy_now", "action_status": "可即刻买入", "strategy_version": "v-test", "strategy_code": BOX_RETEST_4H_STRATEGY, "strategy_signal_id": 42, "strategy_snapshot_json": '{"strategy_code":"box_retest_4h_v1"}', "factor_roles_json": '{"box_breakout_pullback_4h":"trigger"}', "entry_plan": {"entry_action": "可即刻买入", "entry_price": 1.0, "stop_loss": 0.94, "tp1": 1.12}, } payload = _order_payload_from_rec(rec, 1.01, "2026-05-27T00:00:00", {"trade_notional_usdt": 100}) assert payload["strategy_code"] == BOX_RETEST_4H_STRATEGY assert payload["strategy_signal_id"] == 42 result = _open_trade( pg_conn, rec, 1.0, "2026-05-27T00:00:00", config={ "enabled": True, "trade_notional_usdt": 100, "trade_leverage": 1, "account_equity_usdt": 20000, "fee_rate": 0, "min_rec_score": 0, "min_rr": 0, "max_stop_loss_leverage_risk_pct": 999, "max_cumulative_leverage": 999, "max_account_drawdown_pause_pct": 0, "weak_entry_pause": {"enabled": False}, }, push_open_card=False, ) assert result["opened"] is True row = pg_conn.execute("SELECT * FROM paper_trades WHERE id=%s", (result["trade_id"],)).fetchone() event = pg_conn.execute("SELECT * FROM paper_trade_events WHERE trade_id=%s", (result["trade_id"],)).fetchone() assert row["strategy_code"] == BOX_RETEST_4H_STRATEGY assert row["strategy_signal_id"] == 42 assert event["strategy_code"] == BOX_RETEST_4H_STRATEGY assert strategy_label(row["strategy_code"]) == "4H箱体突破回踩"