Chan/crypto_wyckoff/rules/phase_rules.py

"""Phase A–E rules (primarily weekly)."""

from __future__ import annotations

from typing import Any

from crypto_wyckoff.domain_models import WyckoffCycle, WyckoffPhase
from crypto_wyckoff.rules.base import RuleHit, WyckoffRule


def _f(ctx: dict[str, Any], key: str, default: float = 0.0) -> float:
    v = ctx.get("features", {}).get(key, default)
    try:
        return float(v) if v is not None else default
    except (TypeError, ValueError):
        return default


def _cycle(ctx: dict[str, Any]) -> str:
    return (ctx.get("cycle") or {}).get("cycle") or WyckoffCycle.UNKNOWN.value


class PhaseARule(WyckoffRule):
    rule_id = "phase_a"
    category = "phase"
    timeframes = ("1w", "1d")

    def evaluate(self, context: dict[str, Any]) -> RuleHit | None:
        cycle = _cycle(context)
        if cycle not in (WyckoffCycle.ACCUMULATION.value, WyckoffCycle.DISTRIBUTION.value,
                         WyckoffCycle.RE_ACCUMULATION.value, WyckoffCycle.RE_DISTRIBUTION.value):
            return None
        # Stopping action: high vol + large range recently, still range-bound
        vol_ratio = _f(context, "volume_ratio")
        range_last = _f(context, "bar_range_atr")
        if vol_ratio >= 1.4 and range_last >= 1.2:
            return RuleHit(
                rule_id=self.rule_id,
                phase=WyckoffPhase.A.value,
                confidence=70.0,
                score=65.0,
                reasons=["放量宽幅波动，疑似 Phase A 停止行为"],
            )
        return None


class PhaseBRule(WyckoffRule):
    rule_id = "phase_b"
    category = "phase"
    timeframes = ("1w", "1d")

    def evaluate(self, context: dict[str, Any]) -> RuleHit | None:
        cycle = _cycle(context)
        if cycle not in (WyckoffCycle.ACCUMULATION.value, WyckoffCycle.DISTRIBUTION.value):
            return None
        adx = _f(context, "adx")
        range_pct = _f(context, "range_pct_60")
        pos = _f(context, "range_position")  # 0=low 1=high of range
        if adx < 20 and 0.25 < pos < 0.75 and range_pct < 0.30:
            return RuleHit(
                rule_id=self.rule_id,
                phase=WyckoffPhase.B.value,
                confidence=72.0,
                score=68.0,
                reasons=["区间中部震荡，疑似 Phase B 建仓/派发"],
            )
        return None


class PhaseCRule(WyckoffRule):
    rule_id = "phase_c"
    category = "phase"
    timeframes = ("1w", "1d")

    def evaluate(self, context: dict[str, Any]) -> RuleHit | None:
        cycle = _cycle(context)
        pos = _f(context, "range_position")
        spring_like = _f(context, "spring_score_hint")
        utad_like = _f(context, "utad_score_hint")
        if cycle in (WyckoffCycle.ACCUMULATION.value, WyckoffCycle.RE_ACCUMULATION.value):
            if pos < 0.25 or spring_like >= 50:
                return RuleHit(
                    rule_id=self.rule_id,
                    phase=WyckoffPhase.C.value,
                    confidence=75.0 + min(15.0, spring_like * 0.15),
                    score=78.0,
                    reasons=["区间低位测试，疑似 Phase C (Spring/Test)"],
                )
        if cycle in (WyckoffCycle.DISTRIBUTION.value, WyckoffCycle.RE_DISTRIBUTION.value):
            if pos > 0.75 or utad_like >= 50:
                return RuleHit(
                    rule_id=self.rule_id,
                    phase=WyckoffPhase.C.value,
                    confidence=75.0,
                    score=78.0,
                    reasons=["区间高位测试，疑似 Phase C (UTAD)"],
                )
        return None


class PhaseDRule(WyckoffRule):
    rule_id = "phase_d"
    category = "phase"
    timeframes = ("1w", "1d")

    def evaluate(self, context: dict[str, Any]) -> RuleHit | None:
        cycle = _cycle(context)
        close = _f(context, "close")
        ma20 = _f(context, "ma20")
        range_high = _f(context, "range_high")
        range_low = _f(context, "range_low")
        vol_ratio = _f(context, "volume_ratio")
        if cycle in (WyckoffCycle.ACCUMULATION.value, WyckoffCycle.RE_ACCUMULATION.value):
            if close > ma20 and range_high > 0 and close >= range_high * 0.98 and vol_ratio >= 1.1:
                return RuleHit(
                    rule_id=self.rule_id,
                    phase=WyckoffPhase.D.value,
                    confidence=80.0,
                    score=82.0,
                    reasons=["突破区间上沿放量，疑似 Phase D SOS"],
                )
        if cycle in (WyckoffCycle.DISTRIBUTION.value, WyckoffCycle.RE_DISTRIBUTION.value):
            if close < ma20 and range_low > 0 and close <= range_low * 1.02:
                return RuleHit(
                    rule_id=self.rule_id,
                    phase=WyckoffPhase.D.value,
                    confidence=80.0,
                    score=82.0,
                    reasons=["跌破区间下沿，疑似 Phase D SOW"],
                )
        return None


class PhaseERule(WyckoffRule):
    rule_id = "phase_e"
    category = "phase"
    timeframes = ("1w", "1d")

    def evaluate(self, context: dict[str, Any]) -> RuleHit | None:
        cycle = _cycle(context)
        # Markup/Markdown already imply trend continuation (Phase E of prior structure)
        if cycle == WyckoffCycle.MARKUP.value:
            return RuleHit(
                rule_id=self.rule_id,
                phase=WyckoffPhase.E.value,
                confidence=78.0,
                score=80.0,
                reasons=["趋势上行，对应 Phase E Markup"],
            )
        if cycle == WyckoffCycle.MARKDOWN.value:
            return RuleHit(
                rule_id=self.rule_id,
                phase=WyckoffPhase.E.value,
                confidence=78.0,
                score=80.0,
                reasons=["趋势下行，对应 Phase E Markdown"],
            )
        return None


def build_rules() -> list[WyckoffRule]:
    # More specific phases first
    return [PhaseDRule(), PhaseCRule(), PhaseARule(), PhaseBRule(), PhaseERule()]