alphax/app/strategies/altcoin_breakout.py

"""Altcoin breakout strategy profiles.

These builders turn existing evidence into complete strategy signals. A factor
can support a strategy, but the strategy still owns trigger freshness, entry
quality and risk semantics.
"""

from __future__ import annotations

from app.core.factor_roles import CONFIRMATION, ENTRY, PREREQUISITE, RISK, TRIGGER
from app.core.strategy_contract import StrategySignal, current_strategy_version
from app.core.strategy_registry import (
    LONG_MOMENTUM_BREAKOUT_STRATEGY,
    LONG_SECOND_WAVE_PULLBACK_STRATEGY,
)


def _safe_float(value, default=0.0) -> float:
    try:
        if value is None or value == "":
            return default
        return float(value)
    except Exception:
        return default


def _signals_text(result: dict) -> str:
    return " ".join(str(x or "") for x in (result or {}).get("signals") or [])


def _has_short_context(result: dict) -> bool:
    text = _signals_text(result)
    market_context = (result or {}).get("market_context") or {}
    short_ctx = (result or {}).get("short_breakdown_retest_1h") or market_context.get("short_breakdown_retest_1h") or {}
    return bool(short_ctx.get("detected")) or any(
        key in text
        for key in ("破位反抽做空", "等待反抽失败", "反抽失败确认", "breakdown_retest_1h_short")
    )


def _has_second_wave_context(result: dict) -> bool:
    text = _signals_text(result)
    market_context = (result or {}).get("market_context") or {}
    has_first_wave = (
        bool((result or {}).get("top_gainer_24h") or market_context.get("top_gainer_24h"))
        or "24h强势榜" in text
        or "量价齐飞" in text
        or "放量" in text
    )
    has_pullback_context = any(key in text for key in ("回踩", "箱体", "EMA", "前高", "底部抬高", "静K"))
    return bool(has_first_wave and has_pullback_context)


def _trigger_context(result: dict) -> dict:
    return (result or {}).get("trigger_context") or ((result or {}).get("market_context") or {}).get("trigger_context") or {}


def _has_current_trigger(result: dict) -> bool:
    trigger = _trigger_context(result)
    text = _signals_text(result)
    return bool(trigger.get("current_triggers")) or "15min即刻入场" in text or "15min强突破" in text or "15min 强突破" in text


def _status_for_entry(result: dict, entry_plan: dict | None = None, *, require_current_trigger: bool = False, allow_wait: bool = True) -> tuple[str, list[str]]:
    reasons = []
    entry_action = str((entry_plan or {}).get("entry_action") or ((result or {}).get("entry_plan") or {}).get("entry_action") or (result or {}).get("entry_action") or "").strip()
    if require_current_trigger and not _has_current_trigger(result):
        return "observe", ["缺少当前低周期触发"]
    if entry_action in ("可即刻买入", "即刻买入"):
        return "candidate", reasons
    if entry_action == "等回踩" and allow_wait:
        return "candidate", reasons
    if entry_action:
        reasons.append(f"当前入场动作 {entry_action} 不满足策略执行条件")
    return "observe", reasons


def build_long_momentum_breakout_signal(*, symbol: str, result: dict, entry_plan: dict | None = None) -> StrategySignal | None:
    if _has_short_context(result) or _has_second_wave_context(result):
        return None
    text = _signals_text(result)
    has_15m = any(key in text for key in ("15min", "15m", "短周期", "5m/15m"))
    has_1h_participation = "量价齐飞" in text or ("连续" in text and "放量" in text) or ("1H" in text and ("突破" in text or "起爆" in text))
    if not (has_15m and has_1h_participation):
        return None
    status, reasons = _status_for_entry(result, entry_plan, require_current_trigger=True, allow_wait=True)
    score = _safe_float(result.get("score"))
    confidence = min(100.0, max(0.0, score * 7 + 18))
    trigger = {
        "factor_code": "momentum_breakout_15m_1h",
        "factor_label": "15m突破 + 1H放量/波动增强",
        "has_current_trigger": _has_current_trigger(result),
        "trigger_status": _trigger_context(result).get("trigger_status") or "",
        "entry_action": (entry_plan or {}).get("entry_action") or "",
        "opportunity_level": "intraday",
    }
    return StrategySignal(
        strategy_code=LONG_MOMENTUM_BREAKOUT_STRATEGY,
        strategy_version=current_strategy_version(),
        symbol=symbol,
        direction="long",
        status=status,
        confidence=confidence,
        score=score,
        trigger=trigger,
        factor_roles={
            "momentum_breakout_15m_1h": TRIGGER,
            "volume_consecutive_1h": CONFIRMATION,
            "vp_fly_1h_current": CONFIRMATION,
            "breakout_15m_current": ENTRY,
            "pullback_15m_confirm": ENTRY,
            "false_breakout": RISK,
            "risk_reward_bad": RISK,
        },
        entry_plan=entry_plan or {},
        risk_plan={
            "invalid_if": ["15m跌回突破K低点", "1H放量后不能延续", "快速冲高回落", "RR不足"],
            "risk_reasons": reasons,
        },
        decision_log={"module": LONG_MOMENTUM_BREAKOUT_STRATEGY, "decision": status, "reasons": reasons},
    )


def build_long_second_wave_pullback_signal(*, symbol: str, result: dict, entry_plan: dict | None = None) -> StrategySignal | None:
    if _has_short_context(result):
        return None
    text = _signals_text(result)
    if not _has_second_wave_context(result):
        return None
    status, reasons = _status_for_entry(result, entry_plan, require_current_trigger=False, allow_wait=True)
    score = _safe_float(result.get("score"))
    confidence = min(100.0, max(0.0, score * 6 + (12 if "24h强势榜" in text else 0) + (8 if _has_current_trigger(result) else 0)))
    return StrategySignal(
        strategy_code=LONG_SECOND_WAVE_PULLBACK_STRATEGY,
        strategy_version=current_strategy_version(),
        symbol=symbol,
        direction="long",
        status=status,
        confidence=confidence,
        score=score,
        trigger={
            "factor_code": "second_wave_pullback_1h",
            "factor_label": "强势第一波后回踩承接",
            "has_current_trigger": _has_current_trigger(result),
            "trigger_status": _trigger_context(result).get("trigger_status") or "",
            "opportunity_level": "swing_1_3d",
        },
        factor_roles={
            "cex_top_gainer": PREREQUISITE,
            "vp_fly_1h_current": CONFIRMATION,
            "box_breakout_pullback_1h": CONFIRMATION,
            "box_breakout_pullback_4h": CONFIRMATION,
            "second_wave_pullback_1h": TRIGGER,
            "pullback_15m_confirm": ENTRY,
            "false_breakout": RISK,
        },
        entry_plan=entry_plan or {},
        risk_plan={
            "invalid_if": ["跌回第一波启动区", "回踩放量跌破", "高位追涨无承接", "RR不足"],
            "risk_reasons": reasons,
        },
        decision_log={"module": LONG_SECOND_WAVE_PULLBACK_STRATEGY, "decision": status, "reasons": reasons},
    )


def build_volume_ignition_1h_signal(*, symbol: str, result: dict, entry_plan: dict | None = None) -> StrategySignal | None:
    return None


def build_compression_breakout_4h_signal(*, symbol: str, result: dict, entry_plan: dict | None = None) -> StrategySignal | None:
    return None


def build_intraday_momentum_15m_signal(*, symbol: str, result: dict, entry_plan: dict | None = None) -> StrategySignal | None:
    return None