"""Independent short strategy builders. Short setups are not inverted long setups. They need their own trigger, confirmation, invalidation and review lineage. """ from __future__ import annotations from app.core.factor_roles import CONFIRMATION, ENTRY, RISK, TRIGGER from app.core.strategy_contract import StrategySignal, current_strategy_version from app.core.strategy_registry import BREAKDOWN_RETEST_SHORT_1H_STRATEGY, SHORT_WEAK_BOUNCE_FAILURE_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 detect_breakdown_retest_short_1h(df, *, change_24h: float = 0.0) -> dict: """Detect a 1H breakdown followed by a failed retest of the broken box. This is intentionally conservative: the setup needs a prior box, a close below support, a retest near the broken support, and rejection back below it. """ if df is None or len(df) < 60: return {"detected": False, "reason": "insufficient_data"} try: work = df.tail(72).copy() prev = work.iloc[:-8] recent = work.iloc[-8:] current = float(work["close"].iloc[-1]) box_low = float(prev["low"].tail(48).min()) box_high = float(prev["high"].tail(48).max()) if current <= 0 or box_low <= 0 or box_high <= box_low: return {"detected": False, "reason": "invalid_box"} box_width_pct = (box_high / box_low - 1) * 100 if box_width_pct > 38: return {"detected": False, "reason": "box_too_wide", "box_width_pct": round(box_width_pct, 2)} closes = [float(x) for x in recent["close"].tolist()] highs = [float(x) for x in recent["high"].tolist()] opens = [float(x) for x in recent["open"].tolist()] breakdown_closes = [x for x in closes if x < box_low * 0.992] touched_retest = any(box_low * 0.988 <= h <= box_low * 1.035 for h in highs[-5:]) latest_rejected = current < box_low * 0.992 and closes[-1] < opens[-1] below_ema = current < float(work["close"].ewm(span=25, adjust=False).mean().iloc[-1]) if not breakdown_closes or not touched_retest: return { "detected": False, "reason": "no_breakdown_retest", "box_low": round(box_low, 8), "box_high": round(box_high, 8), } stop_level = max(max(highs[-5:]) * 1.012, box_low * 1.025) risk = max(stop_level - current, current * 0.025) target_1 = max(current - risk * 1.8, current * 0.72) quality_score = 0 quality_score += 3 if latest_rejected else 1 quality_score += 2 if below_ema else 0 quality_score += 2 if float(change_24h or 0) <= -3 else 0 quality_score += 1 if box_width_pct <= 24 else 0 quality = "优质" if quality_score >= 7 else "良好" if quality_score >= 5 else "普通" return { "detected": True, "breakdown_level": round(box_low, 8), "box_high": round(box_high, 8), "box_width_pct": round(box_width_pct, 2), "retest_zone": round(box_low, 8), "stop_level": round(stop_level, 8), "target_1": round(target_1, 8), "quality": quality, "quality_score": quality_score, "score": quality_score, "retest_rejected": latest_rejected, "relative_weakness": bool(below_ema or float(change_24h or 0) <= -3), "pullback_age_bars": 0, "signals": [ f"1H破位反抽做空(破位{box_low:.6g})", "反抽失败确认" if latest_rejected else "等待反抽失败确认", ], } except Exception as exc: return {"detected": False, "reason": f"error:{exc}"} def build_breakdown_retest_short_1h_signal( *, symbol: str, current_price: float, detection: dict, entry_plan: dict | None = None, market_regime: dict | None = None, decision_log: dict | None = None, ) -> StrategySignal | None: """Build a short signal for breakdown -> failed retest setups. Expected detection fields are intentionally simple so scanner/confirm implementations can evolve without changing the strategy contract: detected, breakdown_level, retest_zone, stop_level, target_1, quality, retest_rejected, score. """ if not (detection or {}).get("detected"): return None entry_plan = dict(entry_plan or {}) market_regime = market_regime or {} quality = str(detection.get("quality") or "") retest_rejected = bool(detection.get("retest_rejected")) current_price = _safe_float(current_price) retest_zone = _safe_float(detection.get("retest_zone") or detection.get("breakdown_level")) distance_pct = abs(current_price / retest_zone - 1) * 100 if current_price > 0 and retest_zone > 0 else 0.0 risk_level = str(market_regime.get("risk_level") or "medium").lower() reasons = [] status = "candidate" if quality not in {"良好", "优质"}: status = "observe" reasons.append(f"反抽质量 {quality or '未知'}，不直接做空") if not retest_rejected: status = "observe" reasons.append("尚未确认反抽失败") if risk_level in {"low", "medium"} and not bool(detection.get("relative_weakness")): status = "observe" reasons.append("市场并非明显弱势，且缺少相对弱势确认") if distance_pct > 8: status = "observe" reasons.append(f"当前价离反抽区 {distance_pct:.1f}%，不追空") entry_plan.setdefault("side", "short") entry_plan.setdefault("entry_action", "可即刻买入" if status == "candidate" else "观察") entry_plan.setdefault("entry_price", current_price) entry_plan.setdefault("stop_loss", detection.get("stop_level")) entry_plan.setdefault("tp1", detection.get("target_1")) entry_plan.setdefault("risk_reward_ok", True) score = _safe_float(detection.get("score")) confidence = min(100.0, max(0.0, score * 8 + (12 if retest_rejected else 0))) return StrategySignal( strategy_code=BREAKDOWN_RETEST_SHORT_1H_STRATEGY, strategy_version=current_strategy_version(), symbol=symbol, direction="short", status=status, confidence=confidence, score=score, trigger={ "factor_code": "breakdown_retest_1h_short", "factor_label": "1H破位反抽做空", "breakdown_level": detection.get("breakdown_level"), "retest_zone": retest_zone, "stop_level": detection.get("stop_level"), "target_1": detection.get("target_1"), "quality": quality, "retest_rejected": retest_rejected, "distance_to_retest_zone_pct": round(distance_pct, 4), "risk_level": risk_level, }, factor_roles={ "breakdown_retest_1h_short": TRIGGER, "retest_reject_15m_short": ENTRY, "market_risk_off_short": CONFIRMATION, "false_breakout": RISK, "funding_extreme": RISK, }, entry_plan=entry_plan, risk_plan={ "invalid_if": ["重新站回破位区", "反抽放量突破", "BTC/ETH快速转强", "RR不足"], "risk_reasons": reasons, }, decision_log=decision_log or { "module": BREAKDOWN_RETEST_SHORT_1H_STRATEGY, "decision": status, "reasons": reasons, }, ) def build_short_weak_bounce_failure_signal( *, symbol: str, result: dict, entry_plan: dict | None = None, market_regime: dict | None = None, ) -> StrategySignal | None: """Build a short signal for weak bounce failures in risk-off regimes.""" result = result or {} entry_plan = dict(entry_plan or {}) text = " ".join(str(x or "") for x in result.get("signals") or []) market_regime = market_regime or result.get("market_regime") or (result.get("market_context") or {}).get("market_regime") or {} risk_level = str(market_regime.get("risk_level") or "").lower() regime = str(market_regime.get("regime") or "").lower() has_weak_bounce = any(key in text for key in ("弱反弹", "反抽失败", "空头加速", "放量阴线", "破位")) has_entry = any(key in text for key in ("15min", "15m", "反抽失败", "空头加速")) risk_off = risk_level in {"high", "critical"} or regime in {"risk_off", "bearish", "downtrend"} if not (has_weak_bounce and has_entry and risk_off): return None action = str(entry_plan.get("entry_action") or result.get("entry_action") or "").strip() status = "candidate" if action in {"可即刻买入", "即刻买入", "可开空"} else "observe" reasons = [] if status == "candidate" else ["弱反弹失败已出现，但还缺少当前开空动作"] score = _safe_float(result.get("score")) confidence = min(100.0, max(0.0, score * 7 + 12)) entry_plan.setdefault("side", "short") entry_plan.setdefault("entry_action", "可即刻买入" if status == "candidate" else "观察") return StrategySignal( strategy_code=SHORT_WEAK_BOUNCE_FAILURE_STRATEGY, strategy_version=current_strategy_version(), symbol=symbol, direction="short", status=status, confidence=confidence, score=score, trigger={ "factor_code": "weak_bounce_failure_15m_1h_short", "factor_label": "弱势反弹失败做空", "risk_level": risk_level, "regime": regime, "entry_action": action, }, factor_roles={ "weak_bounce_failure_15m_1h_short": TRIGGER, "retest_reject_15m_short": ENTRY, "market_risk_off_short": CONFIRMATION, "funding_extreme": RISK, }, entry_plan=entry_plan, risk_plan={ "invalid_if": ["重新站回反抽高点", "BTC/ETH快速转强", "15m空头结构失效", "RR不足"], "risk_reasons": reasons, }, decision_log={"module": SHORT_WEAK_BOUNCE_FAILURE_STRATEGY, "decision": status, "reasons": reasons}, ) __all__ = [ "build_breakdown_retest_short_1h_signal", "build_short_weak_bounce_failure_signal", "detect_breakdown_retest_short_1h", ]