| """ |
| Chart pattern detection engine. |
| Detects technical analysis patterns on kline data and returns drawing instructions |
| for the frontend to overlay on candlestick charts. |
| """ |
|
|
| import logging |
| from dataclasses import dataclass, field |
|
|
| import numpy as np |
|
|
| from app.models.schemas import KlineData |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| @dataclass |
| class PatternResult: |
| pattern_type: str |
| name: str |
| direction: str |
| confidence: float |
| start_idx: int |
| end_idx: int |
| description: str |
| draw_lines: list[dict] = field(default_factory=list) |
|
|
| def to_dict(self) -> dict: |
| return { |
| "pattern_type": self.pattern_type, |
| "name": self.name, |
| "direction": self.direction, |
| "confidence": round(self.confidence, 1), |
| "start_idx": self.start_idx, |
| "end_idx": self.end_idx, |
| "description": self.description, |
| "draw_lines": self.draw_lines, |
| } |
|
|
|
|
| PATTERN_REGISTRY: dict[str, dict] = { |
| "trendline_support": {"name": "支撑趋势线", "description": "连接多个低点形成上升/下降支撑线"}, |
| "trendline_resistance": {"name": "阻力趋势线", "description": "连接多个高点形成上升/下降阻力线"}, |
| "ascending_triangle": {"name": "上升三角形", "description": "水平阻力 + 上升支撑，看涨突破形态"}, |
| "descending_triangle": {"name": "下降三角形", "description": "水平支撑 + 下降阻力，看跌突破形态"}, |
| "symmetric_triangle": {"name": "对称三角形", "description": "收敛三角形，突破方向待定"}, |
| "double_top": {"name": "双顶 (M头)", "description": "两个近似高点，看跌反转形态"}, |
| "double_bottom": {"name": "双底 (W底)", "description": "两个近似低点，看涨反转形态"}, |
| "head_shoulders": {"name": "头肩顶", "description": "头部高于两肩，看跌反转形态"}, |
| "inv_head_shoulders": {"name": "头肩底", "description": "头部低于两肩，看涨反转形态"}, |
| "channel_up": {"name": "上升通道", "description": "平行的上升支撑和阻力"}, |
| "channel_down": {"name": "下降通道", "description": "平行的下降支撑和阻力"}, |
| "wedge_rising": {"name": "上升楔形", "description": "收窄的上升通道，看跌"}, |
| "wedge_falling": {"name": "下降楔形", "description": "收窄的下降通道，看涨"}, |
| "support_level": {"name": "水平支撑", "description": "价格多次触及的水平支撑位"}, |
| "resistance_level": {"name": "水平阻力", "description": "价格多次触及的水平阻力位"}, |
| } |
|
|
|
|
| def _find_local_extrema(prices: np.ndarray, order: int = 5) -> tuple[list[int], list[int]]: |
| highs, lows = [], [] |
| for i in range(order, len(prices) - order): |
| if all(prices[i] >= prices[i - j] for j in range(1, order + 1)) and \ |
| all(prices[i] >= prices[i + j] for j in range(1, order + 1)): |
| highs.append(i) |
| if all(prices[i] <= prices[i - j] for j in range(1, order + 1)) and \ |
| all(prices[i] <= prices[i + j] for j in range(1, order + 1)): |
| lows.append(i) |
| return highs, lows |
|
|
|
|
| def _fit_line(x_indices: list[int], y_values: list[float]) -> tuple[float, float]: |
| if len(x_indices) < 2: |
| return 0.0, y_values[0] if y_values else 0.0 |
| x = np.array(x_indices, dtype=float) |
| y = np.array(y_values, dtype=float) |
| slope, intercept = np.polyfit(x, y, 1) |
| return float(slope), float(intercept) |
|
|
|
|
| def _make_line(x1: int, y1: float, x2: int, y2: float, color: str, style: str = "solid", label: str = "") -> dict: |
| return {"x1": x1, "y1": round(y1, 2), "x2": x2, "y2": round(y2, 2), |
| "color": color, "style": style, "label": label} |
|
|
|
|
| def detect_trendlines(highs_arr: np.ndarray, lows_arr: np.ndarray, n: int, order: int = 5) -> list[PatternResult]: |
| results = [] |
| high_idxs, low_idxs = _find_local_extrema(highs_arr, order) |
| _, low_idxs2 = _find_local_extrema(lows_arr, order) |
|
|
| if len(low_idxs2) >= 2: |
| vals = [float(lows_arr[i]) for i in low_idxs2] |
| slope, intercept = _fit_line(low_idxs2, vals) |
| y1 = intercept + slope * low_idxs2[0] |
| extend_y2 = intercept + slope * (n - 1) |
| direction = "bullish" if slope > 0 else "bearish" if slope < 0 else "neutral" |
| results.append(PatternResult( |
| pattern_type="trendline_support", name="支撑趋势线", |
| direction=direction, confidence=65 + min(len(low_idxs2), 5) * 5, |
| start_idx=low_idxs2[0], end_idx=n - 1, |
| description=f"连接{len(low_idxs2)}个低点，斜率{'上升' if slope > 0 else '下降'}", |
| draw_lines=[_make_line(low_idxs2[0], y1, n - 1, extend_y2, "#3fb950", "dashed", "支撑线")], |
| )) |
|
|
| if len(high_idxs) >= 2: |
| vals = [float(highs_arr[i]) for i in high_idxs] |
| slope, intercept = _fit_line(high_idxs, vals) |
| y1 = intercept + slope * high_idxs[0] |
| extend_y2 = intercept + slope * (n - 1) |
| direction = "bullish" if slope > 0 else "bearish" |
| results.append(PatternResult( |
| pattern_type="trendline_resistance", name="阻力趋势线", |
| direction=direction, confidence=65 + min(len(high_idxs), 5) * 5, |
| start_idx=high_idxs[0], end_idx=n - 1, |
| description=f"连接{len(high_idxs)}个高点，斜率{'上升' if slope > 0 else '下降'}", |
| draw_lines=[_make_line(high_idxs[0], y1, n - 1, extend_y2, "#f85149", "dashed", "阻力线")], |
| )) |
|
|
| return results |
|
|
|
|
| def detect_triangles(highs_arr: np.ndarray, lows_arr: np.ndarray, n: int) -> list[PatternResult]: |
| results = [] |
| high_idxs, low_idxs = _find_local_extrema(highs_arr, 5) |
| _, low_idxs2 = _find_local_extrema(lows_arr, 5) |
|
|
| if len(high_idxs) >= 2 and len(low_idxs2) >= 2: |
| h_slope, h_int = _fit_line(high_idxs, [float(highs_arr[i]) for i in high_idxs]) |
| l_slope, l_int = _fit_line(low_idxs2, [float(lows_arr[i]) for i in low_idxs2]) |
|
|
| h_y1 = h_int + h_slope * high_idxs[0] |
| h_y2 = h_int + h_slope * high_idxs[-1] |
| l_y1 = l_int + l_slope * low_idxs2[0] |
| l_y2 = l_int + l_slope * low_idxs2[-1] |
|
|
| h_range = abs(h_y2 - h_y1) |
| l_range = abs(l_y2 - l_y1) |
| avg_price = float(np.mean(highs_arr)) |
| threshold = avg_price * 0.003 |
|
|
| if h_range < threshold and l_slope > 0: |
| results.append(PatternResult( |
| pattern_type="ascending_triangle", name="上升三角形", |
| direction="bullish", confidence=72, |
| start_idx=min(high_idxs[0], low_idxs2[0]), end_idx=max(high_idxs[-1], low_idxs2[-1]), |
| description="水平阻力线 + 上升支撑线，看涨突破", |
| draw_lines=[ |
| _make_line(high_idxs[0], h_y1, high_idxs[-1], h_y2, "#f85149", "solid", "水平阻力"), |
| _make_line(low_idxs2[0], l_y1, low_idxs2[-1], l_y2, "#3fb950", "solid", "上升支撑"), |
| ], |
| )) |
| elif l_range < threshold and h_slope < 0: |
| results.append(PatternResult( |
| pattern_type="descending_triangle", name="下降三角形", |
| direction="bearish", confidence=72, |
| start_idx=min(high_idxs[0], low_idxs2[0]), end_idx=max(high_idxs[-1], low_idxs2[-1]), |
| description="下降阻力线 + 水平支撑线，看跌突破", |
| draw_lines=[ |
| _make_line(high_idxs[0], h_y1, high_idxs[-1], h_y2, "#f85149", "solid", "下降阻力"), |
| _make_line(low_idxs2[0], l_y1, low_idxs2[-1], l_y2, "#3fb950", "solid", "水平支撑"), |
| ], |
| )) |
| elif h_slope < 0 and l_slope > 0: |
| results.append(PatternResult( |
| pattern_type="symmetric_triangle", name="对称三角形", |
| direction="neutral", confidence=68, |
| start_idx=min(high_idxs[0], low_idxs2[0]), end_idx=max(high_idxs[-1], low_idxs2[-1]), |
| description="收敛三角形，等待方向突破", |
| draw_lines=[ |
| _make_line(high_idxs[0], h_y1, high_idxs[-1], h_y2, "#f85149", "solid", "下降阻力"), |
| _make_line(low_idxs2[0], l_y1, low_idxs2[-1], l_y2, "#3fb950", "solid", "上升支撑"), |
| ], |
| )) |
| return results |
|
|
|
|
| def detect_double_patterns(highs_arr: np.ndarray, lows_arr: np.ndarray, closes: np.ndarray, n: int) -> list[PatternResult]: |
| results = [] |
| high_idxs, _ = _find_local_extrema(highs_arr, 5) |
| _, low_idxs = _find_local_extrema(lows_arr, 5) |
|
|
| if len(high_idxs) >= 2: |
| for i in range(len(high_idxs) - 1): |
| h1, h2 = high_idxs[i], high_idxs[i + 1] |
| if h2 - h1 < 10: |
| continue |
| p1, p2 = float(highs_arr[h1]), float(highs_arr[h2]) |
| avg_p = (p1 + p2) / 2 |
| if abs(p1 - p2) / avg_p < 0.01: |
| neckline = float(np.min(lows_arr[h1:h2 + 1])) |
| results.append(PatternResult( |
| pattern_type="double_top", name="双顶 (M头)", |
| direction="bearish", confidence=75, |
| start_idx=h1, end_idx=h2, |
| description=f"两个高点 {p1:.2f}/{p2:.2f}，颈线 {neckline:.2f}", |
| draw_lines=[ |
| _make_line(h1, p1, h2, p2, "#f85149", "solid", "双顶连线"), |
| _make_line(h1, neckline, h2, neckline, "#d29922", "dashed", "颈线"), |
| ], |
| )) |
| break |
|
|
| if len(low_idxs) >= 2: |
| for i in range(len(low_idxs) - 1): |
| l1, l2 = low_idxs[i], low_idxs[i + 1] |
| if l2 - l1 < 10: |
| continue |
| p1, p2 = float(lows_arr[l1]), float(lows_arr[l2]) |
| avg_p = (p1 + p2) / 2 |
| if abs(p1 - p2) / avg_p < 0.01: |
| neckline = float(np.max(highs_arr[l1:l2 + 1])) |
| results.append(PatternResult( |
| pattern_type="double_bottom", name="双底 (W底)", |
| direction="bullish", confidence=75, |
| start_idx=l1, end_idx=l2, |
| description=f"两个低点 {p1:.2f}/{p2:.2f}，颈线 {neckline:.2f}", |
| draw_lines=[ |
| _make_line(l1, p1, l2, p2, "#3fb950", "solid", "双底连线"), |
| _make_line(l1, neckline, l2, neckline, "#d29922", "dashed", "颈线"), |
| ], |
| )) |
| break |
|
|
| return results |
|
|
|
|
| def detect_support_resistance(highs_arr: np.ndarray, lows_arr: np.ndarray, n: int) -> list[PatternResult]: |
| results = [] |
| price_range = float(np.max(highs_arr) - np.min(lows_arr)) |
| tolerance = price_range * 0.008 |
|
|
| high_idxs, _ = _find_local_extrema(highs_arr, 4) |
| _, low_idxs = _find_local_extrema(lows_arr, 4) |
|
|
| resistance_levels = [] |
| for i in high_idxs: |
| p = float(highs_arr[i]) |
| found = False |
| for lvl in resistance_levels: |
| if abs(p - lvl["price"]) < tolerance: |
| lvl["touches"] += 1 |
| lvl["indices"].append(i) |
| lvl["price"] = (lvl["price"] + p) / 2 |
| found = True |
| break |
| if not found: |
| resistance_levels.append({"price": p, "touches": 1, "indices": [i]}) |
|
|
| for lvl in resistance_levels: |
| if lvl["touches"] >= 2: |
| results.append(PatternResult( |
| pattern_type="resistance_level", name="水平阻力", |
| direction="bearish", confidence=60 + lvl["touches"] * 8, |
| start_idx=lvl["indices"][0], end_idx=n - 1, |
| description=f"阻力位 {lvl['price']:.2f}，触及 {lvl['touches']} 次", |
| draw_lines=[_make_line(lvl["indices"][0], lvl["price"], n - 1, lvl["price"], "#f85149", "dotted", f"阻力 {lvl['price']:.2f}")], |
| )) |
|
|
| support_levels = [] |
| for i in low_idxs: |
| p = float(lows_arr[i]) |
| found = False |
| for lvl in support_levels: |
| if abs(p - lvl["price"]) < tolerance: |
| lvl["touches"] += 1 |
| lvl["indices"].append(i) |
| lvl["price"] = (lvl["price"] + p) / 2 |
| found = True |
| break |
| if not found: |
| support_levels.append({"price": p, "touches": 1, "indices": [i]}) |
|
|
| for lvl in support_levels: |
| if lvl["touches"] >= 2: |
| results.append(PatternResult( |
| pattern_type="support_level", name="水平支撑", |
| direction="bullish", confidence=60 + lvl["touches"] * 8, |
| start_idx=lvl["indices"][0], end_idx=n - 1, |
| description=f"支撑位 {lvl['price']:.2f}，触及 {lvl['touches']} 次", |
| draw_lines=[_make_line(lvl["indices"][0], lvl["price"], n - 1, lvl["price"], "#3fb950", "dotted", f"支撑 {lvl['price']:.2f}")], |
| )) |
|
|
| return results |
|
|
|
|
| def detect_channels(highs_arr: np.ndarray, lows_arr: np.ndarray, n: int) -> list[PatternResult]: |
| results = [] |
| high_idxs, _ = _find_local_extrema(highs_arr, 5) |
| _, low_idxs = _find_local_extrema(lows_arr, 5) |
|
|
| if len(high_idxs) >= 2 and len(low_idxs) >= 2: |
| h_slope, h_int = _fit_line(high_idxs, [float(highs_arr[i]) for i in high_idxs]) |
| l_slope, l_int = _fit_line(low_idxs, [float(lows_arr[i]) for i in low_idxs]) |
|
|
| slope_diff = abs(h_slope - l_slope) |
| avg_slope = (abs(h_slope) + abs(l_slope)) / 2 |
|
|
| if avg_slope > 0 and slope_diff / avg_slope < 0.4: |
| start = min(high_idxs[0], low_idxs[0]) |
| end = max(high_idxs[-1], low_idxs[-1]) |
| h_y1, h_y2 = h_int + h_slope * start, h_int + h_slope * end |
| l_y1, l_y2 = l_int + l_slope * start, l_int + l_slope * end |
|
|
| if h_slope > 0 and l_slope > 0: |
| ptype, pname, direction = "channel_up", "上升通道", "bullish" |
| elif h_slope < 0 and l_slope < 0: |
| ptype, pname, direction = "channel_down", "下降通道", "bearish" |
| else: |
| return results |
|
|
| width_start = abs(h_y1 - l_y1) |
| width_end = abs(h_y2 - l_y2) |
| if width_end < width_start * 0.7: |
| ptype = "wedge_rising" if h_slope > 0 else "wedge_falling" |
| pname = "上升楔形" if h_slope > 0 else "下降楔形" |
| direction = "bearish" if h_slope > 0 else "bullish" |
|
|
| results.append(PatternResult( |
| pattern_type=ptype, name=pname, |
| direction=direction, confidence=70, |
| start_idx=start, end_idx=end, |
| description=f"{pname}，通道宽度 {width_start:.2f} → {width_end:.2f}", |
| draw_lines=[ |
| _make_line(start, h_y1, end, h_y2, "#f85149", "solid", "上轨"), |
| _make_line(start, l_y1, end, l_y2, "#3fb950", "solid", "下轨"), |
| ], |
| )) |
|
|
| return results |
|
|
|
|
| def detect_patterns(klines: list[KlineData], enabled_patterns: list[str] | None = None) -> list[dict]: |
| if len(klines) < 30: |
| return [] |
|
|
| closes = np.array([k.close for k in klines]) |
| highs = np.array([k.high for k in klines]) |
| lows = np.array([k.low for k in klines]) |
| n = len(klines) |
|
|
| all_patterns: list[PatternResult] = [] |
|
|
| all_types = enabled_patterns or list(PATTERN_REGISTRY.keys()) |
|
|
| need_trendline = any(t.startswith("trendline") for t in all_types) |
| need_triangle = any(t.endswith("triangle") for t in all_types) |
| need_double = any(t.startswith("double") for t in all_types) |
| need_sr = any(t.endswith("level") for t in all_types) |
| need_channel = any(t.startswith("channel") or t.startswith("wedge") for t in all_types) |
|
|
| if need_trendline: |
| all_patterns.extend(detect_trendlines(highs, lows, n)) |
| if need_triangle: |
| all_patterns.extend(detect_triangles(highs, lows, n)) |
| if need_double: |
| all_patterns.extend(detect_double_patterns(highs, lows, closes, n)) |
| if need_sr: |
| all_patterns.extend(detect_support_resistance(highs, lows, n)) |
| if need_channel: |
| all_patterns.extend(detect_channels(highs, lows, n)) |
|
|
| filtered = [p for p in all_patterns if p.pattern_type in all_types] |
| filtered.sort(key=lambda p: p.confidence, reverse=True) |
|
|
| return [p.to_dict() for p in filtered] |
|
|
|
|
| def get_pattern_registry() -> list[dict]: |
| return [{"type": k, **v} for k, v in PATTERN_REGISTRY.items()] |
|
|
