819 lines
24 KiB
Python
819 lines
24 KiB
Python
"""入场信号分类 + 供需分析 + 量价形态识别
|
||
|
||
五种入场信号类型：
|
||
- 突破型 (breakout): 放量突破 20 日阻力位
|
||
- 确认型 (breakout_confirm): 突破后 1-2 日放量确认
|
||
- 回踩型 (pullback): 上升趋势中缩量回踩均线支撑
|
||
- 启动型 (launch): 缩量整理后首日放量
|
||
- 反转型 (reversal): 连续下跌后放量长阳反转
|
||
"""
|
||
|
||
import logging
|
||
from enum import Enum
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
from app.config import settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class EntrySignal(str, Enum):
|
||
BREAKOUT = "breakout"
|
||
BREAKOUT_CONFIRM = "breakout_confirm"
|
||
PULLBACK = "pullback"
|
||
LAUNCH = "launch"
|
||
REVERSAL = "reversal"
|
||
NONE = "none"
|
||
|
||
|
||
# ── 入场信号检测 ──
|
||
|
||
|
||
def classify_entry_signal(df: pd.DataFrame) -> dict:
|
||
"""分类入场信号类型
|
||
|
||
优先级: breakout > breakout_confirm > pullback > launch > reversal
|
||
|
||
Returns:
|
||
{
|
||
"signal_type": EntrySignal,
|
||
"signal_score": float (0-100),
|
||
"details": dict,
|
||
}
|
||
"""
|
||
if df is None or len(df) < 30:
|
||
return {"signal_type": EntrySignal.NONE, "signal_score": 0, "details": {}}
|
||
|
||
# 突破型
|
||
result = detect_breakout(df)
|
||
if result:
|
||
result["signal_score"] = _score_breakout(df, result)
|
||
return result
|
||
|
||
# 突破确认型（突破后 1-3 日放量确认）
|
||
result = detect_breakout_confirm(df)
|
||
if result:
|
||
result["signal_score"] = _score_breakout_confirm(df, result)
|
||
return result
|
||
|
||
# 回踩型
|
||
result = detect_pullback(df)
|
||
if result:
|
||
result["signal_score"] = _score_pullback(df, result)
|
||
return result
|
||
|
||
# 启动型（缩量横盘后首日放量）
|
||
result = detect_launch(df)
|
||
if result:
|
||
result["signal_score"] = _score_launch(df, result)
|
||
return result
|
||
|
||
# 反转型（连续下跌后放量长阳）
|
||
result = detect_reversal(df)
|
||
if result:
|
||
result["signal_score"] = _score_reversal(df, result)
|
||
return result
|
||
|
||
return {"signal_type": EntrySignal.NONE, "signal_score": 0, "details": {}}
|
||
|
||
|
||
def detect_breakout(df: pd.DataFrame) -> dict | None:
|
||
"""突破型检测（严格版：必须真正突破 + 显著放量）
|
||
|
||
条件：
|
||
1. 收盘价真正突破 20 日高点（close > resist_20d），不再允许"贴近"
|
||
2. 量能显著放大 (vol > vol_ma5 * 1.5)
|
||
3. MA5 > MA20（短期偏多）
|
||
4. 收盘偏强 (close 在当日上半部)
|
||
"""
|
||
last = df.iloc[-1]
|
||
|
||
required = ["close", "high", "low", "vol", "ma5", "ma10", "ma20", "vol_ma5"]
|
||
if any(c not in df.columns for c in required):
|
||
return None
|
||
if pd.isna(last["vol_ma5"]) or last["vol_ma5"] == 0:
|
||
return None
|
||
|
||
if len(df) < 22:
|
||
return None
|
||
resist_20d = df["high"].iloc[-21:-1].max()
|
||
|
||
# 严格：必须真正突破（收盘价 > 20日高点），不允许"贴近未突破"
|
||
if last["close"] <= resist_20d:
|
||
return None
|
||
|
||
# 量能显著放大（从 1.2 提高到 1.5）
|
||
if last["vol"] <= last["vol_ma5"] * 1.5:
|
||
return None
|
||
|
||
# MA5 > MA20
|
||
if pd.isna(last["ma20"]) or not (last["ma5"] > last["ma20"]):
|
||
return None
|
||
|
||
# 收盘偏强
|
||
day_range = last["high"] - last["low"]
|
||
if day_range > 0 and (last["close"] - last["low"]) / day_range < 0.45:
|
||
return None
|
||
|
||
breakout_pct = (last["close"] - resist_20d) / resist_20d * 100
|
||
|
||
return {
|
||
"signal_type": EntrySignal.BREAKOUT,
|
||
"details": {
|
||
"resist_level": round(resist_20d, 2),
|
||
"resistance_price": round(resist_20d, 2), # 阻力位价格，供止损计算用
|
||
"breakout_pct": round(breakout_pct, 2),
|
||
"volume_ratio": round(last["vol"] / last["vol_ma5"], 2),
|
||
},
|
||
}
|
||
|
||
|
||
def detect_breakout_confirm(df: pd.DataFrame) -> dict | None:
|
||
"""突破确认型检测
|
||
|
||
条件：近 1-3 日内曾突破 20 日高点，今日放量确认（不要求再创新高）
|
||
这捕捉了"突破→缩量回踩确认→再放量"的经典模式。
|
||
|
||
1. 近 3 日内有 1 日 close > 20日前的高点
|
||
2. 今日 vol > vol_ma5（放量确认）
|
||
3. 今日 close > 昨日 close * 0.99（没有大幅回落）
|
||
4. MA5 > MA20
|
||
"""
|
||
if len(df) < 24:
|
||
return None
|
||
|
||
last = df.iloc[-1]
|
||
required = ["close", "high", "low", "vol", "pct_chg", "ma5", "ma20", "vol_ma5"]
|
||
if any(c not in df.columns for c in required):
|
||
return None
|
||
if pd.isna(last["vol_ma5"]) or last["vol_ma5"] == 0:
|
||
return None
|
||
if pd.isna(last["ma20"]):
|
||
return None
|
||
|
||
# 20 日阻力（不含近 3 日）
|
||
resist_20d = df["high"].iloc[-23:-3].max()
|
||
|
||
# 近 3 日内有突破
|
||
recent_3 = df.tail(3)
|
||
had_breakout = any(
|
||
row["close"] > resist_20d
|
||
for _, row in recent_3.iterrows()
|
||
if not pd.isna(row["close"])
|
||
)
|
||
if not had_breakout:
|
||
return None
|
||
|
||
# 今日放量
|
||
if last["vol"] <= last["vol_ma5"] * 1.0:
|
||
return None
|
||
|
||
# 今日没有大幅回落
|
||
if last["close"] < df.iloc[-2]["close"] * 0.99:
|
||
return None
|
||
|
||
# MA5 > MA20
|
||
if not (last["ma5"] > last["ma20"]):
|
||
return None
|
||
|
||
return {
|
||
"signal_type": EntrySignal.BREAKOUT_CONFIRM,
|
||
"details": {
|
||
"resist_level": round(resist_20d, 2),
|
||
"volume_ratio": round(last["vol"] / last["vol_ma5"], 2),
|
||
"confirm_pct": round(last["pct_chg"], 2),
|
||
},
|
||
}
|
||
|
||
|
||
def detect_pullback(df: pd.DataFrame) -> dict | None:
|
||
"""回踩型检测
|
||
|
||
条件：
|
||
1. MA20 > MA60 — 中长期上升趋势
|
||
2. 近 5 日内有 MA5 > MA20 — 近期处于上升趋势
|
||
3. 价格靠近 MA10 或 MA20 (<3%)
|
||
4. 近 3 日缩量（相对前 5 日）或 当日开始放量反弹
|
||
5. RSI 30-70（放宽）
|
||
"""
|
||
last = df.iloc[-1]
|
||
|
||
required = ["close", "vol", "ma5", "ma10", "ma20", "ma60", "rsi14"]
|
||
if any(c not in df.columns for c in required):
|
||
return None
|
||
if pd.isna(last["ma60"]) or pd.isna(last["rsi14"]):
|
||
return None
|
||
|
||
# 中长期上升趋势
|
||
if not (last["ma20"] > last["ma60"]):
|
||
return None
|
||
|
||
# 近 5 日内有 MA5 > MA20
|
||
recent_5 = df.tail(5)
|
||
has_uptrend = any(
|
||
row["ma5"] > row["ma20"]
|
||
for _, row in recent_5.iterrows()
|
||
if not pd.isna(row["ma5"]) and not pd.isna(row["ma20"])
|
||
)
|
||
if not has_uptrend:
|
||
return None
|
||
|
||
# 价格靠近 MA10 或 MA20（放宽至 3%）
|
||
near_ma10 = (
|
||
not pd.isna(last["ma10"])
|
||
and last["ma10"] > 0
|
||
and abs(last["close"] - last["ma10"]) / last["ma10"] < 0.03
|
||
)
|
||
near_ma20 = (
|
||
not pd.isna(last["ma20"])
|
||
and last["ma20"] > 0
|
||
and abs(last["close"] - last["ma20"]) / last["ma20"] < 0.03
|
||
)
|
||
if not (near_ma10 or near_ma20):
|
||
return None
|
||
|
||
# 缩量回调 OR 当日放量反弹（二选一）
|
||
if len(df) < 8:
|
||
return None
|
||
recent_3_vol = df["vol"].tail(3).mean()
|
||
prev_5_vol = df["vol"].iloc[-8:-3].mean()
|
||
|
||
shrinking = prev_5_vol > 0 and recent_3_vol / prev_5_vol < 0.9
|
||
bouncing_today = (
|
||
last["pct_chg"] > 0
|
||
and not pd.isna(last["vol_ma5"])
|
||
and last["vol_ma5"] > 0
|
||
and last["vol"] > last["vol_ma5"] * 1.1
|
||
)
|
||
|
||
if not (shrinking or bouncing_today):
|
||
return None
|
||
|
||
# RSI（放宽至 30-70）
|
||
if not (30 <= last["rsi14"] <= 70):
|
||
return None
|
||
|
||
support_ma = "MA10" if near_ma10 else "MA20"
|
||
support_price = last["ma10"] if near_ma10 else last["ma20"]
|
||
shrink_ratio = recent_3_vol / prev_5_vol if prev_5_vol > 0 else 1
|
||
|
||
return {
|
||
"signal_type": EntrySignal.PULLBACK,
|
||
"details": {
|
||
"support_ma": support_ma,
|
||
"support_price": round(support_price, 2),
|
||
"volume_shrink_ratio": round(shrink_ratio, 2),
|
||
"bouncing": bouncing_today,
|
||
},
|
||
}
|
||
|
||
|
||
def detect_launch(df: pd.DataFrame) -> dict | None:
|
||
"""启动型检测（缩量横盘后首日放量）
|
||
|
||
条件：
|
||
1. 近 5-10 日缩量至均量的 60% 以下
|
||
2. 今日放量（vol > vol_ma5 * 1.2）
|
||
3. close > MA60（中长线偏多）
|
||
4. 今日涨跌幅 > 0（方向向上）
|
||
"""
|
||
last = df.iloc[-1]
|
||
|
||
required = ["close", "high", "vol", "pct_chg", "ma5", "ma20", "ma60", "vol_ma5"]
|
||
if any(c not in df.columns for c in required):
|
||
return None
|
||
if pd.isna(last["ma60"]) or pd.isna(last["vol_ma5"]) or last["vol_ma5"] == 0:
|
||
return None
|
||
|
||
# close > MA60
|
||
if last["close"] < last["ma60"]:
|
||
return None
|
||
|
||
# 今日放量
|
||
vol_ratio_today = last["vol"] / last["vol_ma5"]
|
||
if vol_ratio_today < 1.2:
|
||
return None
|
||
|
||
# 今日上涨
|
||
if last["pct_chg"] <= 0:
|
||
return None
|
||
|
||
# 近 5-10 日有缩量期（存在 vol < vol_ma5 * 0.6 的日子）
|
||
if len(df) < 12:
|
||
return None
|
||
recent_10 = df.iloc[-11:-1] # 不含今日
|
||
vol_ma_col = "vol_ma10" if "vol_ma10" in df.columns else "vol_ma5"
|
||
had_dry_up = False
|
||
for _, row in recent_10.iterrows():
|
||
if pd.isna(row.get(vol_ma_col)) or row[vol_ma_col] == 0:
|
||
continue
|
||
if row["vol"] < row[vol_ma_col] * 0.6:
|
||
had_dry_up = True
|
||
break
|
||
|
||
if not had_dry_up:
|
||
return None
|
||
|
||
# 近 10 日振幅不大（横盘整理）
|
||
recent_10_close = df["close"].iloc[-11:-1]
|
||
if len(recent_10_close) < 5:
|
||
return None
|
||
price_range = (recent_10_close.max() - recent_10_close.min()) / recent_10_close.min()
|
||
if price_range > 0.12: # 12% 以内算横盘
|
||
return None
|
||
|
||
high_20d = df["high"].tail(20).max()
|
||
|
||
return {
|
||
"signal_type": EntrySignal.LAUNCH,
|
||
"details": {
|
||
"volume_ratio_today": round(vol_ratio_today, 2),
|
||
"price_range_pct": round(price_range * 100, 2),
|
||
"resist_level": round(high_20d, 2),
|
||
},
|
||
}
|
||
|
||
|
||
def detect_reversal(df: pd.DataFrame) -> dict | None:
|
||
"""反转型检测（连续下跌后放量长阳）
|
||
|
||
条件：
|
||
1. 近 5 日有 3 日以上下跌
|
||
2. 今日放量长阳（pct_chg > 3%, vol > vol_ma5 * 1.5）
|
||
3. 收盘价高于近 5 日最高价（强势反转）
|
||
4. RSI 从低位回升（近 5 日最低 RSI < 40）
|
||
"""
|
||
if len(df) < 10:
|
||
return None
|
||
|
||
last = df.iloc[-1]
|
||
required = ["close", "high", "low", "vol", "pct_chg", "rsi14", "vol_ma5"]
|
||
if any(c not in df.columns for c in required):
|
||
return None
|
||
if pd.isna(last["vol_ma5"]) or last["vol_ma5"] == 0 or pd.isna(last["rsi14"]):
|
||
return None
|
||
|
||
# 近 5 日有 3 日以上下跌
|
||
recent_5 = df.tail(5)
|
||
down_count = sum(1 for _, r in recent_5.iterrows() if r["pct_chg"] < 0)
|
||
if down_count < 3:
|
||
return None
|
||
|
||
# 今日放量长阳
|
||
if last["pct_chg"] < 3:
|
||
return None
|
||
if last["vol"] < last["vol_ma5"] * 1.5:
|
||
return None
|
||
|
||
# 收盘价高于近 5 日最高价（强势反转）
|
||
high_5d = df["high"].iloc[-6:-1].max()
|
||
if last["close"] < high_5d:
|
||
return None
|
||
|
||
# RSI 曾在低位
|
||
rsi_5d = df["rsi14"].tail(5)
|
||
min_rsi = rsi_5d.min()
|
||
if not pd.isna(min_rsi) and min_rsi > 40:
|
||
return None
|
||
|
||
return {
|
||
"signal_type": EntrySignal.REVERSAL,
|
||
"details": {
|
||
"down_days": down_count,
|
||
"reversal_pct": round(last["pct_chg"], 2),
|
||
"volume_ratio": round(last["vol"] / last["vol_ma5"], 2),
|
||
"min_rsi": round(float(min_rsi), 1),
|
||
},
|
||
}
|
||
|
||
|
||
# ── 信号质量评分 ──
|
||
|
||
|
||
def _score_breakout(df: pd.DataFrame, signal: dict) -> float:
|
||
"""突破型评分 (0-100)"""
|
||
last = df.iloc[-1]
|
||
score = 0
|
||
|
||
breakout_pct = signal["details"].get("breakout_pct", 0)
|
||
if breakout_pct > 3:
|
||
score += 25
|
||
elif breakout_pct > 1.5:
|
||
score += 18
|
||
elif breakout_pct > 0:
|
||
score += 12
|
||
# 不再有"贴近未突破"分支，只有真正突破才到这里
|
||
|
||
vol_ratio = signal["details"].get("volume_ratio", 1)
|
||
if vol_ratio > 2.5:
|
||
score += 25
|
||
elif vol_ratio > 2.0:
|
||
score += 20
|
||
elif vol_ratio > 1.5:
|
||
score += 14
|
||
elif vol_ratio > 1.2:
|
||
score += 8
|
||
|
||
if all(c in df.columns for c in ["ma60"]):
|
||
if not pd.isna(last["ma60"]) and last["ma5"] > last["ma10"] > last["ma20"] > last["ma60"]:
|
||
score += 25
|
||
elif last["ma5"] > last["ma10"] > last["ma20"]:
|
||
score += 15
|
||
else:
|
||
score += 10
|
||
|
||
if all(c in df.columns for c in ["dif", "dea"]):
|
||
if not pd.isna(last["dif"]) and last["dif"] > last["dea"] and last["dif"] > 0:
|
||
score += 15
|
||
elif not pd.isna(last["dif"]) and last["dif"] > last["dea"]:
|
||
score += 8
|
||
|
||
if "rsi14" in df.columns and not pd.isna(last["rsi14"]):
|
||
if 50 <= last["rsi14"] <= 75:
|
||
score += 10
|
||
elif 45 <= last["rsi14"] <= 80:
|
||
score += 5
|
||
|
||
return min(score, 100)
|
||
|
||
|
||
def _score_breakout_confirm(df: pd.DataFrame, signal: dict) -> float:
|
||
"""突破确认型评分 (0-100)"""
|
||
last = df.iloc[-1]
|
||
score = 0
|
||
|
||
vol_ratio = signal["details"].get("volume_ratio", 1)
|
||
if vol_ratio > 2.0:
|
||
score += 25
|
||
elif vol_ratio > 1.5:
|
||
score += 18
|
||
elif vol_ratio > 1.0:
|
||
score += 10
|
||
|
||
confirm_pct = signal["details"].get("confirm_pct", 0)
|
||
if confirm_pct > 3:
|
||
score += 20
|
||
elif confirm_pct > 1:
|
||
score += 14
|
||
elif confirm_pct > 0:
|
||
score += 8
|
||
|
||
if all(c in df.columns for c in ["ma60"]):
|
||
if not pd.isna(last["ma60"]) and last["ma5"] > last["ma10"] > last["ma20"] > last["ma60"]:
|
||
score += 25
|
||
elif last["ma5"] > last["ma10"] > last["ma20"]:
|
||
score += 15
|
||
|
||
if all(c in df.columns for c in ["dif", "dea"]):
|
||
if not pd.isna(last["dif"]) and last["dif"] > last["dea"] and last["dif"] > 0:
|
||
score += 15
|
||
elif not pd.isna(last["dif"]) and last["dif"] > last["dea"]:
|
||
score += 8
|
||
|
||
if "rsi14" in df.columns and not pd.isna(last["rsi14"]):
|
||
if 50 <= last["rsi14"] <= 70:
|
||
score += 15
|
||
elif 45 <= last["rsi14"] <= 75:
|
||
score += 8
|
||
|
||
return min(score, 100)
|
||
|
||
|
||
def _score_pullback(df: pd.DataFrame, signal: dict) -> float:
|
||
"""回踩型评分 (0-100)"""
|
||
last = df.iloc[-1]
|
||
score = 0
|
||
|
||
shrink = signal["details"].get("volume_shrink_ratio", 1)
|
||
bouncing = signal["details"].get("bouncing", False)
|
||
if bouncing:
|
||
score += 20 # 已开始反弹加分
|
||
elif shrink < 0.5:
|
||
score += 30
|
||
elif shrink < 0.6:
|
||
score += 22
|
||
elif shrink < 0.7:
|
||
score += 15
|
||
elif shrink < 0.9:
|
||
score += 8
|
||
|
||
support_ma = signal["details"].get("support_ma", "")
|
||
if support_ma == "MA20":
|
||
score += 18
|
||
elif support_ma == "MA10":
|
||
score += 12
|
||
|
||
if all(c in df.columns for c in ["ma60"]):
|
||
if not pd.isna(last["ma60"]) and last["ma5"] > last["ma10"] > last["ma20"] > last["ma60"]:
|
||
score += 25
|
||
elif last["ma5"] > last["ma10"] > last["ma20"]:
|
||
score += 15
|
||
|
||
if all(c in df.columns for c in ["dif", "dea"]):
|
||
if not pd.isna(last["dif"]) and last["dif"] > 0:
|
||
score += 10
|
||
elif not pd.isna(last["dif"]) and last["dif"] > last["dea"]:
|
||
score += 5
|
||
|
||
if "rsi14" in df.columns and not pd.isna(last["rsi14"]):
|
||
if 40 <= last["rsi14"] <= 55:
|
||
score += 15
|
||
elif 35 <= last["rsi14"] <= 65:
|
||
score += 8
|
||
|
||
return min(score, 100)
|
||
|
||
|
||
def _score_launch(df: pd.DataFrame, signal: dict) -> float:
|
||
"""启动型评分 (0-100)"""
|
||
last = df.iloc[-1]
|
||
score = 0
|
||
details = signal.get("details", {})
|
||
|
||
range_pct = details.get("price_range_pct", 10)
|
||
if range_pct < 4:
|
||
score += 20
|
||
elif range_pct < 8:
|
||
score += 14
|
||
elif range_pct < 12:
|
||
score += 8
|
||
|
||
vol_ratio = details.get("volume_ratio_today", 1)
|
||
if vol_ratio > 2.5:
|
||
score += 25
|
||
elif vol_ratio > 1.8:
|
||
score += 18
|
||
elif vol_ratio > 1.2:
|
||
score += 10
|
||
|
||
if all(c in df.columns for c in ["ma60"]):
|
||
if not pd.isna(last["ma60"]) and last["ma5"] > last["ma20"] > last["ma60"]:
|
||
score += 20
|
||
elif last["ma5"] > last["ma20"]:
|
||
score += 10
|
||
|
||
if all(c in df.columns for c in ["dif", "dea"]):
|
||
if not pd.isna(last["dif"]) and last["dif"] > 0:
|
||
score += 15
|
||
elif not pd.isna(last["dif"]):
|
||
score += 5
|
||
|
||
# MACD 金叉或 DIF 上翘加分
|
||
if "dif" in df.columns and "dea" in df.columns and len(df) >= 3:
|
||
if not pd.isna(last["dif"]) and not pd.isna(df.iloc[-3]["dif"]):
|
||
if last["dif"] > df.iloc[-3]["dif"]: # DIF 上翘
|
||
score += 10
|
||
|
||
return min(score, 100)
|
||
|
||
|
||
def _score_reversal(df: pd.DataFrame, signal: dict) -> float:
|
||
"""反转型评分 (0-100)"""
|
||
last = df.iloc[-1]
|
||
score = 0
|
||
|
||
reversal_pct = signal["details"].get("reversal_pct", 0)
|
||
if reversal_pct > 5:
|
||
score += 25
|
||
elif reversal_pct > 3:
|
||
score += 18
|
||
|
||
vol_ratio = signal["details"].get("volume_ratio", 1)
|
||
if vol_ratio > 3:
|
||
score += 25
|
||
elif vol_ratio > 2:
|
||
score += 18
|
||
elif vol_ratio > 1.5:
|
||
score += 12
|
||
|
||
min_rsi = signal["details"].get("min_rsi", 50)
|
||
if min_rsi < 25:
|
||
score += 20 # 极度超卖反转
|
||
elif min_rsi < 30:
|
||
score += 15
|
||
elif min_rsi < 35:
|
||
score += 10
|
||
|
||
# 站上 MA5 加分
|
||
if "ma5" in df.columns and not pd.isna(last["ma5"]):
|
||
if last["close"] > last["ma5"]:
|
||
score += 10
|
||
|
||
# 量能超过近 20 日最大量
|
||
if len(df) >= 20:
|
||
vol_20d_max = df["vol"].iloc[-21:-1].max()
|
||
if last["vol"] > vol_20d_max:
|
||
score += 10
|
||
|
||
return min(score, 100)
|
||
|
||
|
||
# ── 供需分析评分 ──
|
||
|
||
|
||
def score_supply_demand(df: pd.DataFrame) -> float:
|
||
"""供需评分 (0-100)，窗口从 10 日扩展到 20 日
|
||
|
||
维度：
|
||
- 上涨日 vs 下跌日量比 (40): 需求主导度
|
||
- 量能收缩/扩张 (25): 供需变化趋势
|
||
- 量价配合 (20): 上涨放量+下跌缩量
|
||
- 经典量价形态 (15): 缩量横盘→放量、底部量堆
|
||
"""
|
||
score = 0
|
||
|
||
if len(df) < 10:
|
||
return 50.0
|
||
|
||
# 使用 20 日窗口（如果数据够）
|
||
window = min(20, len(df))
|
||
recent = df.tail(window)
|
||
last = df.iloc[-1]
|
||
|
||
# 上涨日 vs 下跌日量比 (40)
|
||
up_days = recent[recent["pct_chg"] > 0]
|
||
down_days = recent[recent["pct_chg"] <= 0]
|
||
|
||
if len(up_days) > 0 and len(down_days) > 0:
|
||
avg_up_vol = up_days["vol"].mean()
|
||
avg_down_vol = down_days["vol"].mean()
|
||
if avg_down_vol > 0:
|
||
ratio = avg_up_vol / avg_down_vol
|
||
if ratio > 2.0:
|
||
score += 40
|
||
elif ratio > 1.5:
|
||
score += 30
|
||
elif ratio > 1.2:
|
||
score += 20
|
||
elif ratio > 1.0:
|
||
score += 10
|
||
elif len(up_days) > len(down_days):
|
||
score += 30
|
||
|
||
# 量能收缩/扩张 (25)
|
||
vol_ma_col = "vol_ma10" if "vol_ma10" in df.columns else "vol_ma5"
|
||
if vol_ma_col in df.columns and not pd.isna(last[vol_ma_col]) and last[vol_ma_col] > 0:
|
||
recent_3_vol = recent["vol"].tail(3).mean()
|
||
shrink_ratio = recent_3_vol / last[vol_ma_col]
|
||
|
||
if shrink_ratio < 0.4:
|
||
score += 25
|
||
elif shrink_ratio < 0.6:
|
||
score += 18
|
||
elif shrink_ratio < 0.8:
|
||
score += 10
|
||
elif shrink_ratio > 1.5:
|
||
score += 20
|
||
elif shrink_ratio > 1.2:
|
||
score += 12
|
||
|
||
# 量价配合 (20)
|
||
if _check_bullish_volume_divergence(df):
|
||
score += 20
|
||
|
||
# 经典量价形态 (15)
|
||
if _detect_volume_pattern(df) > 0:
|
||
score += 15
|
||
|
||
return min(score, 100)
|
||
|
||
|
||
def _detect_volume_pattern(df: pd.DataFrame) -> int:
|
||
"""检测经典量价形态
|
||
|
||
返回形态强度 (0-2):
|
||
1 = 缩量横盘后首日放量
|
||
2 = 底部量堆（近 20 日出现量堆+价格企稳）
|
||
"""
|
||
if len(df) < 20:
|
||
return 0
|
||
|
||
strength = 0
|
||
|
||
# 形态 1：缩量横盘后首日放量
|
||
vol_ma_col = "vol_ma10" if "vol_ma10" in df.columns else "vol_ma5"
|
||
if vol_ma_col in df.columns:
|
||
recent_10 = df.iloc[-11:-1]
|
||
had_dry_up = any(
|
||
row["vol"] < row[vol_ma_col] * 0.5
|
||
for _, row in recent_10.iterrows()
|
||
if not pd.isna(row[vol_ma_col]) and row[vol_ma_col] > 0
|
||
)
|
||
last = df.iloc[-1]
|
||
vol_expanding = (
|
||
not pd.isna(last[vol_ma_col])
|
||
and last[vol_ma_col] > 0
|
||
and last["vol"] > last[vol_ma_col] * 1.3
|
||
and last["pct_chg"] > 0
|
||
)
|
||
if had_dry_up and vol_expanding:
|
||
strength = max(strength, 1)
|
||
|
||
# 形态 2：底部量堆
|
||
if len(df) >= 20:
|
||
vol_20d = df["vol"].tail(20)
|
||
vol_mean = vol_20d.mean()
|
||
if vol_mean > 0:
|
||
vol_max = vol_20d.max()
|
||
vol_min = vol_20d.min()
|
||
# 量堆：存在量能是均量 2 倍以上的日子，且价格没有大幅下跌
|
||
high_vol_days = sum(1 for v in vol_20d if v > vol_mean * 2)
|
||
price_change = (df["close"].iloc[-1] - df["close"].iloc[-20]) / df["close"].iloc[-20] * 100
|
||
if high_vol_days >= 2 and price_change > -3:
|
||
strength = max(strength, 2)
|
||
|
||
return strength
|
||
|
||
|
||
def analyze_volume_pattern(df: pd.DataFrame) -> dict:
|
||
"""量价模式分析"""
|
||
default = {"demand_supply_ratio": 1.0, "volume_trend": "stable", "accumulation_score": 0}
|
||
|
||
if len(df) < 10:
|
||
return default
|
||
|
||
window = min(20, len(df))
|
||
recent = df.tail(window)
|
||
|
||
up_days = recent[recent["pct_chg"] > 0]
|
||
down_days = recent[recent["pct_chg"] <= 0]
|
||
avg_up_vol = up_days["vol"].mean() if len(up_days) > 0 else 0
|
||
avg_down_vol = down_days["vol"].mean() if len(down_days) > 0 else 1
|
||
ratio = round(avg_up_vol / max(avg_down_vol, 1), 2)
|
||
|
||
if len(df) >= 10:
|
||
half = window // 2
|
||
vol_first = df["vol"].iloc[-window:-half].mean()
|
||
vol_recent = df["vol"].tail(half).mean()
|
||
if vol_first > 0:
|
||
trend_ratio = vol_recent / vol_first
|
||
trend = "expanding" if trend_ratio > 1.2 else ("contracting" if trend_ratio < 0.8 else "stable")
|
||
else:
|
||
trend = "stable"
|
||
else:
|
||
trend = "stable"
|
||
|
||
accumulation = 0
|
||
if ratio > 1.5:
|
||
accumulation += 40
|
||
if trend == "expanding" and df.iloc[-1]["pct_chg"] > 0:
|
||
accumulation += 30
|
||
if _check_volume_dry_up_before_rally(df):
|
||
accumulation += 30
|
||
|
||
return {
|
||
"demand_supply_ratio": ratio,
|
||
"volume_trend": trend,
|
||
"accumulation_score": accumulation,
|
||
}
|
||
|
||
|
||
# ── 辅助函数 ──
|
||
|
||
|
||
def _check_higher_highs(df: pd.DataFrame) -> bool:
|
||
if len(df) < 20:
|
||
return False
|
||
recent = df.tail(20)
|
||
highs = recent["high"].values
|
||
first_half_max = highs[:10].max()
|
||
second_half_max = highs[10:].max()
|
||
return second_half_max >= first_half_max
|
||
|
||
|
||
def _check_bullish_volume_divergence(df: pd.DataFrame) -> bool:
|
||
if len(df) < 6:
|
||
return False
|
||
recent = df.tail(5)
|
||
up_days = recent[recent["pct_chg"] > 0]
|
||
down_days = recent[recent["pct_chg"] <= 0]
|
||
if len(up_days) == 0 or len(down_days) == 0:
|
||
return len(up_days) > len(down_days)
|
||
avg_up_vol = up_days["vol"].mean()
|
||
avg_down_vol = down_days["vol"].mean()
|
||
return avg_up_vol > avg_down_vol
|
||
|
||
|
||
def _check_volume_dry_up_before_rally(df: pd.DataFrame) -> bool:
|
||
if len(df) < 15:
|
||
return False
|
||
last = df.iloc[-1]
|
||
vol_ma_col = "vol_ma10" if "vol_ma10" in df.columns else "vol_ma5"
|
||
if vol_ma_col not in df.columns:
|
||
return False
|
||
recent_10 = df.tail(10)
|
||
had_dry_up = any(
|
||
row["vol"] < row[vol_ma_col] * 0.5
|
||
for _, row in recent_10.iterrows()
|
||
if not pd.isna(row[vol_ma_col]) and row[vol_ma_col] > 0
|
||
)
|
||
vol_ma_col_5 = "vol_ma5" if "vol_ma5" in df.columns else vol_ma_col
|
||
vol_expanding = (
|
||
not pd.isna(last[vol_ma_col_5])
|
||
and last[vol_ma_col_5] > 0
|
||
and last["vol"] > last[vol_ma_col_5] * 1.2
|
||
)
|
||
return had_dry_up and vol_expanding
|