alphax/tests/test_opportunity_level.py

import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))

from app.core.opportunity_level import (
    attach_opportunity_level,
    classify_opportunity_level,
    level_tp_parameters,
    select_level_stop_loss,
)
from app.core.opportunity_lifecycle import apply_entry_quality_gate
from app.db.altcoin_db import create_recommendation, get_conn, init_db


def test_intraday_breakout_requires_current_low_timeframe_trigger():
    meta = classify_opportunity_level(
        signals=["1H 量价齐飞K(量3.2x)", "🟢 15min即刻入场信号"],
        entry_plan={"entry_action": "即刻买入"},
        m30_aligned=True,
    )

    assert meta["opportunity_level"] == "intraday_breakout"
    assert meta["label"] == "日内启动"
    assert meta["max_action"] == "buy_now"


def test_short_swing_uses_mid_timeframe_confirmation():
    meta = classify_opportunity_level(
        signals=["1H 量价齐飞K(量3.0x)", "30min 4阳动K(与1H共振)", "4H需求区反弹"],
        entry_plan={"entry_action": "等回踩"},
        m30_aligned=True,
    )

    assert meta["opportunity_level"] == "short_swing"
    assert meta["holding_horizon"] == "1-3天"


def test_higher_timeframe_background_stays_structure_watch():
    meta = classify_opportunity_level(
        signals=["日线 底部缩量(0.6x)", "日线 晨星反转", "1H历史放量阳线已过期(10小时前)"],
        entry_plan={"entry_action": "等回踩"},
    )

    assert meta["opportunity_level"] == "structure_watch"
    assert meta["max_action"] == "wait_pullback"


def test_theme_without_price_trigger_is_research_trend():
    meta = classify_opportunity_level(
        signals=["生态主题扩散", "舆情催化"],
        entry_plan={"entry_action": "观察"},
        sector_context={"hot_sectors": ["AI"]},
    )

    assert meta["opportunity_level"] == "theme_trend"
    assert meta["max_action"] == "observe"


def test_top_gainer_with_current_trigger_becomes_momentum_watch():
    meta = classify_opportunity_level(
        signals=["24h强势榜异动(14.0%,成交额18.0M)", "🟢 15min即刻入场信号", "日线 站稳突破位+36.0%"],
        entry_plan={"entry_action": "即刻买入"},
        market_context={"change_24h": 14.0},
    )

    assert meta["opportunity_level"] == "momentum_watch"
    assert meta["label"] == "强势观察"
    assert meta["max_action"] == "observe"
    assert "24h强势榜异动" in meta["plan_basis"]


def test_level_stop_and_tp_models_are_different():
    stops = [90, 94, 96]
    intraday_stop, _ = select_level_stop_loss(level="intraday_breakout", price=100, entry_price=100, stop_candidates=stops)
    structure_stop, _ = select_level_stop_loss(level="structure_watch", price=100, entry_price=100, stop_candidates=stops)

    assert intraday_stop == 96
    assert structure_stop == 90
    assert level_tp_parameters("intraday_breakout")["tp1_floor"] < level_tp_parameters("structure_watch")["tp1_floor"]


def test_quality_gate_keeps_momentum_watch_observe_not_buy_now():
    meta = classify_opportunity_level(
        signals=["24h强势榜异动(14.0%,成交额18.0M)", "🟢 15min即刻入场信号", "日线 站稳突破位+36.0%"],
        entry_plan={"entry_action": "即刻买入"},
        market_context={"change_24h": 14.0},
    )
    plan = attach_opportunity_level(
        {
            "entry_action": "即刻买入",
            "entry_price": 1.0,
            "stop_loss": 0.94,
            "tp1": 1.1,
            "risk_reward_ok": True,
            "rr1": 1.6,
        },
        meta,
    )

    action, gated_plan, reasons = apply_entry_quality_gate(
        action_status="可即刻买入",
        entry_plan=plan,
        signals=["24h强势榜异动(14.0%,成交额18.0M)", "🟢 15min即刻入场信号", "日线 站稳突破位+36.0%"],
        current_price=1.0,
        market_context={"change_24h": 14.0},
    )

    assert action == "观察"
    assert gated_plan["opportunity_level"] == "momentum_watch"
    assert any("强势观察" in reason for reason in reasons)


def test_quality_gate_caps_structure_watch_without_current_trigger():
    meta = classify_opportunity_level(
        signals=["日线 需求区反弹", "4H静K蓄力观察(4静K)"],
        entry_plan={"entry_action": "即刻买入"},
    )
    plan = attach_opportunity_level(
        {
            "entry_action": "即刻买入",
            "entry_price": 1.0,
            "stop_loss": 0.92,
            "tp1": 1.16,
            "risk_reward_ok": True,
            "rr1": 2.0,
        },
        meta,
    )

    action, gated_plan, reasons = apply_entry_quality_gate(
        action_status="可即刻买入",
        entry_plan=plan,
        signals=["日线 需求区反弹", "4H静K蓄力观察(4静K)"],
        current_price=1.0,
        market_context={"change_24h": 2.0},
    )

    assert action != "可即刻买入"
    assert gated_plan["opportunity_level"] == "structure_watch"
    assert any("结构观察" in reason for reason in reasons)


def test_create_recommendation_persists_opportunity_level_fields():
    init_db()
    plan = attach_opportunity_level(
        {
            "entry_action": "即刻买入",
            "entry_price": 1.0,
            "stop_loss": 0.95,
            "tp1": 1.08,
            "tp2": 1.12,
            "risk_reward_ok": True,
            "rr1": 1.6,
            "entry_trigger_confirmed": True,
        },
        classify_opportunity_level(
            signals=["1H 量价齐飞K(量3.2x)", "🟢 15min即刻入场信号"],
            entry_plan={"entry_action": "即刻买入"},
            m30_aligned=True,
        ),
    )

    rec_id = create_recommendation(
        symbol="TEST/USDT",
        rec_state="爆发",
        rec_score=18,
        entry_price=plan["entry_price"],
        stop_loss=plan["stop_loss"],
        tp1=plan["tp1"],
        tp2=plan["tp2"],
        signals=["1H 量价齐飞K(量3.2x)", "🟢 15min即刻入场信号"],
        entry_plan=plan,
    )

    conn = get_conn()
    try:
        row = conn.execute(
            "SELECT opportunity_level, opportunity_level_label, holding_horizon, entry_plan_json FROM recommendation WHERE id=%s",
            (rec_id,),
        ).fetchone()
    finally:
        conn.close()

    stored_plan = json.loads(row["entry_plan_json"])
    assert row["opportunity_level"] == "intraday_breakout"
    assert row["opportunity_level_label"] == "日内启动"
    assert row["holding_horizon"] == "数小时-1天"
    assert stored_plan["entry_model"] == "15m触发 / 1H突破延续"