mahmoud20138 Skill: Price Action | Domain: trading | Category: analysis | Level: intermediate Tags:
trading,analysis,price-action,harmonics,elliott-wave,technical-analysis
Price Action Pure Engine
Price Action Pure Engine — No Indicators
import pandas as pd, numpy as np
from scipy.signal import argrelextrema
class PriceActionEngine:
@staticmethod
def key_level_reaction(df: pd.DataFrame, levels: list[float], tolerance_atr_mult: float = 0.3) -> list[dict]:
"""Detect price reactions at key levels — the core of PA trading."""
atr = (df["high"] - df["low"]).rolling(14).mean()
reactions = []
for level in levels:
recent = df.tail(20)
for i, (idx, bar) in enumerate(recent.iterrows()):
tol = atr.loc[idx] * tolerance_atr_mult
touching = bar["low"] <= level + tol and bar["high"] >= level - tol
if touching:
body = abs(bar["close"] - bar["open"])
lower_wick = min(bar["open"], bar["close"]) - bar["low"]
upper_wick = bar["high"] - max(bar["open"], bar["close"])
if lower_wick > body * 2 and bar["close"] > bar["open"]:
reactions.append({"level": level, "time": idx, "type": "bullish_rejection",
"signal": "BUY — rejection pin bar at key level"})
elif upper_wick > body * 2 and bar["close"] < bar["open"]:
reactions.append({"level": level, "time": idx, "type": "bearish_rejection",
"signal": "SELL — rejection pin bar at key level"})
elif bar["close"] > level + tol and bar["open"] < level:
reactions.append({"level": level, "time": idx, "type": "bullish_engulf_level",
"signal": "BUY — bullish engulfing through key level"})
return reactions
@staticmethod
def inside_bar_breakout(df: pd.DataFrame) -> list[dict]:
"""Inside bar = compression before expansion. Trade the breakout."""
signals = []
for i in range(1, min(20, len(df))):
idx = len(df) - i
mother = df.iloc[idx - 1]
inside = df.iloc[idx]
if inside["high"] < mother["high"] and inside["low"] > mother["low"]:
if idx + 1 < len(df):
breakout = df.iloc[idx + 1]
if breakout["close"] > mother["high"]:
signals.append({"type": "inside_bar_bullish_breakout", "idx": idx,
"entry": round(mother["high"], 5), "stop": round(mother["low"], 5)})
elif breakout["close"] < mother["low"]:
signals.append({"type": "inside_bar_bearish_breakout", "idx": idx,
"entry": round(mother["low"], 5), "stop": round(mother["high"], 5)})
else:
signals.append({"type": "inside_bar_forming", "idx": idx,
"buy_trigger": round(mother["high"], 5),
"sell_trigger": round(mother["low"], 5)})
return signals
@staticmethod
def engulfing_at_structure(df: pd.DataFrame, order: int = 10) -> list[dict]:
"""Engulfing candles at swing highs/lows — highest probability PA setup."""
highs = argrelextrema(df["high"].values, np.greater, order=order)[0]
lows = argrelextrema(df["low"].values, np.less, order=order)[0]
signals = []
for i in range(1, min(10, len(df))):
idx = len(df) - i
curr = df.iloc[idx]
prev = df.iloc[idx - 1]
# Bullish engulfing near swing low
near_low = any(abs(df["low"].iloc[l] - curr["low"]) < (df["high"] - df["low"]).rolling(14).mean().iloc[idx] for l in lows if abs(l - idx) < 20)
if curr["close"] > curr["open"] and prev["close"] < prev["open"] and curr["close"] > prev["open"] and curr["open"] < prev["close"] and near_low:
signals.append({"type": "bullish_engulfing_at_structure", "idx": idx, "signal": "A+ BUY"})
# Bearish engulfing near swing high
near_high = any(abs(df["high"].iloc[h] - curr["high"]) < (df["high"] - df["low"]).rolling(14).mean().iloc[idx] for h in highs if abs(h - idx) < 20)
if curr["close"] < curr["open"] and prev["close"] > prev["open"] and curr["open"] > prev["close"] and curr["close"] < prev["open"] and near_high:
signals.append({"type": "bearish_engulfing_at_structure", "idx": idx, "signal": "A+ SELL"})
return signals
@staticmethod
def full_pa_scan(df: pd.DataFrame, key_levels: list[float] = None) -> dict:
levels = key_levels or []
return {
"level_reactions": PriceActionEngine.key_level_reaction(df, levels) if levels else [],
"inside_bars": PriceActionEngine.inside_bar_breakout(df),
"engulfing_at_structure": PriceActionEngine.engulfing_at_structure(df),
"principle": "Trade what you SEE, not what you think. PA at key levels = highest probability.",
}
Candlestick Pattern Vision
Candlestick Pattern Vision
Overview
Pure computer vision approach to candlestick pattern detection. Extracts individual candle geometries from chart images via contour detection, then classifies patterns using geometric ratios. Works on any chart screenshot — TradingView, MT5, phone captures.
Stack
- OpenCV 4.13 — contour detection, morphological analysis, connected components
- scikit-image 0.26 — region properties, label analysis
- numpy 2.4 — geometric computations
1. Candle Geometry Extractor
import cv2
import numpy as np
from skimage import measure, morphology as sk_morphology
from dataclasses import dataclass
from typing import Optional
@dataclass
class CandleGeometry:
"""Geometric properties of a single candlestick extracted from image."""
x_center: int # horizontal position (pixel)
y_top: int # highest point (wick top)
y_bottom: int # lowest point (wick bottom)
body_top: int # body top (max of open/close)
body_bottom: int # body bottom (min of open/close)
width: int # body width
is_bullish: bool # green/white = bullish
confidence: float # detection confidence
@property
def total_height(self) -> int:
return self.y_bottom - self.y_top
@property
def body_height(self) -> int:
return self.body_bottom - self.body_top
@property
def upper_wick(self) -> int:
return self.body_top - self.y_top
@property
def lower_wick(self) -> int:
return self.y_bottom - self.body_bottom
@property
def body_ratio(self) -> float:
"""Body size relative to total candle."""
return self.body_height / max(self.total_height, 1)
@property
def upper_wick_ratio(self) -> float:
return self.upper_wick / max(self.total_height, 1)
@property
def lower_wick_ratio(self) -> float:
return self.lower_wick / max(self.total_height, 1)
class CandleExtractor:
"""Extract individual candlestick geometries from a preprocessed chart image."""
@staticmethod
def extract_candles(img: np.ndarray, color_info: dict = None) -> list[CandleGeometry]:
"""
Extract all candlesticks from a chart image.
Uses color segmentation + contour analysis + connected components.
"""
hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
h, w = img.shape[:2]
# Separate bullish (green) and bearish (red) candles
green_mask = cv2.inRange(hsv, (35, 30, 30), (85, 255, 255))
red_mask1 = cv2.inRange(hsv, (0, 30, 30), (15, 255, 255))
red_mask2 = cv2.inRange(hsv, (165, 30, 30), (180, 255, 255))
red_mask = cv2.bitwise_or(red_mask1, red_mask2)
candles = []
for mask, is_bull in [(green_mask, True), (red_mask, False)]:
# Morphological cleanup
kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (3, 3))
mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, kernel, iterations=2)
mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, kernel, iterations=1)
# Connected components (scikit-image 0.26)
labels = measure.label(mask, connectivity=2)
regions = measure.regionprops(labels)
for region in regions:
# Filter by size — candles have specific aspect ratios
bbox = region.bbox # (min_row, min_col, max_row, max_col)
region_h = bbox[2] - bbox[0]
region_w = bbox[3] - bbox[1]
if region_h < 5 or region_w < 2: # Too small
continue
if region_w > w * 0.1: # Too wide (probably not a candle)
continue
if region.area < 20: # Too few pixels
continue
# Determine body vs wick
# Body is the thickest part; wick is thin
col_slice = mask[bbox[0]:bbox[2], bbox[1]:bbox[3]]
row_widths = np.sum(col_slice > 0, axis=1)
# Body rows: where width is > 50% of max width
max_width = row_widths.max()
body_rows = np.where(row_widths > max_width * 0.5)[0]
if len(body_rows) > 0:
body_top_local = body_rows[0]
body_bottom_local = body_rows[-1]
else:
body_top_local = 0
body_bottom_local = region_h
candles.append(CandleGeometry(
x_center=int(region.centroid[1]),
y_top=bbox[0],
y_bottom=bbox[2],
body_top=bbox[0] + body_top_local,
body_bottom=bbox[0] + body_bottom_local,
width=region_w,
is_bullish=is_bull,
confidence=min(region.area / 100, 1.0),
))
# Sort by x position (left to right = chronological)
candles.sort(key=lambda c: c.x_center)
return candles
2. Single Candle Pattern Classifier
class SingleCandleClassifier:
"""Classify individual candlestick patterns from geometry."""
@staticmethod
def classify(candle: CandleGeometry) -> dict:
br = candle.body_ratio
uwr = candle.upper_wick_ratio
lwr = candle.lower_wick_ratio
patterns = []
# Doji: very small body
if br < 0.1:
if uwr > 0.3 and lwr > 0.3:
patterns.append({"pattern": "long_legged_doji", "bias": "reversal", "strength": 0.7})
elif uwr > 0.4:
patterns.append({"pattern": "gravestone_doji", "bias": "bearish_reversal", "strength": 0.75})
elif lwr > 0.4:
patterns.append({"pattern": "dragonfly_doji", "bias": "bullish_reversal", "strength": 0.75})
else:
patterns.append({"pattern": "doji", "bias": "indecision", "strength": 0.5})
# Hammer / Hanging Man: small body at top, long lower wick
elif br < 0.35 and lwr > 0.55 and uwr < 0.1:
if candle.is_bullish:
patterns.append({"pattern": "hammer", "bias": "bullish_reversal", "strength": 0.8})
else:
patterns.append({"pattern": "hanging_man", "bias": "bearish_reversal", "strength": 0.7})
# Inverted Hammer / Shooting Star: small body at bottom, long upper wick
elif br < 0.35 and uwr > 0.55 and lwr < 0.1:
if candle.is_bullish:
patterns.append({"pattern": "inverted_hammer", "bias": "bullish_reversal", "strength": 0.65})
else:
patterns.append({"pattern": "shooting_star", "bias": "bearish_reversal", "strength": 0.8})
# Marubozu: full body, no wicks
elif br > 0.85 and uwr < 0.05 and lwr < 0.05:
bias = "strong_bullish" if candle.is_bullish else "strong_bearish"
patterns.append({"pattern": "marubozu", "bias": bias, "strength": 0.85})
# Pin bar: small body, one very long wick
elif br < 0.25 and (uwr > 0.6 or lwr > 0.6):
direction = "bullish" if lwr > uwr else "bearish"
patterns.append({"pattern": "pin_bar", "bias": f"{direction}_reversal", "strength": 0.8})
# Spinning top: small body, both wicks present
elif br < 0.3 and uwr > 0.2 and lwr > 0.2:
patterns.append({"pattern": "spinning_top", "bias": "indecision", "strength": 0.4})
if not patterns:
patterns.append({"pattern": "regular", "bias": "bullish" if candle.is_bullish else "bearish", "strength": 0.3})
return {"candle": patterns[0], "all_matches": patterns,
"geometry": {"body_ratio": round(br, 3), "upper_wick_ratio": round(uwr, 3), "lower_wick_ratio": round(lwr, 3)}}
3. Multi-Candle Pattern Classifier
class MultiCandleClassifier:
"""Detect patterns spanning 2-3 candles."""
@staticmethod
def classify_sequence(candles: list[CandleGeometry]) -> list[dict]:
patterns = []
if len(candles) < 2:
return patterns
for i in range(len(candles) - 1):
c1, c2 = candles[i], candles[i + 1]
# Bullish Engulfing
if not c1.is_bullish and c2.is_bullish and c2.body_height > c1.body_height * 1.2:
patterns.append({"pattern": "bullish_engulfing", "position": i, "bias": "bullish", "strength": 0.82})
# Bearish Engulfing
if c1.is_bullish and not c2.is_bullish and c2.body_height > c1.body_height * 1.2:
patterns.append({"pattern": "bearish_engulfing", "position": i, "bias": "bearish", "strength": 0.82})
# Tweezer Top/Bottom
if abs(c1.y_top - c2.y_top) < 3: # Equal highs
patterns.append({"pattern": "tweezer_top", "position": i, "bias": "bearish", "strength": 0.65})
if abs(c1.y_bottom - c2.y_bottom) < 3: # Equal lows
patterns.append({"pattern": "tweezer_bottom", "position": i, "bias": "bullish", "strength": 0.65})
# 3-candle patterns
for i in range(len(candles) - 2):
c1, c2, c3 = candles[i], candles[i + 1], candles[i + 2]
# Morning Star
if not c1.is_bullish and c2.body_ratio < 0.2 and c3.is_bullish and c3.body_height > c1.body_height * 0.5:
patterns.append({"pattern": "morning_star", "position": i, "bias": "bullish_reversal", "strength": 0.85})
# Evening Star
if c1.is_bullish and c2.body_ratio < 0.2 and not c3.is_bullish and c3.body_height > c1.body_height * 0.5:
patterns.append({"pattern": "evening_star", "position": i, "bias": "bearish_reversal", "strength": 0.85})
# Three White Soldiers
if all(c.is_bullish for c in [c1, c2, c3]) and c2.body_bottom < c1.body_bottom and c3.body_bottom < c2.body_bottom:
patterns.append({"pattern": "three_white_soldiers", "position": i, "bias": "strong_bullish", "strength": 0.8})
# Three Black Crows
if all(not c.is_bullish for c in [c1, c2, c3]) and c2.body_top > c1.body_top and c3.body_top > c2.body_top:
patterns.append({"pattern": "three_black_crows", "position": i, "bias": "strong_bearish", "strength": 0.8})
return patterns
@staticmethod
def full_scan(img: np.ndarray) -> dict:
"""Complete candlestick pattern scan from image."""
candles = CandleExtractor.extract_candles(img)
single_patterns = [SingleCandleClassifier.classify(c) for c in candles[-10:]] # Last 10
multi_patterns = MultiCandleClassifier.classify_sequence(candles[-10:])
return {
"candles_detected": len(candles),
"last_candle": single_patterns[-1] if single_patterns else None,
"recent_single_patterns": [p["candle"] for p in single_patterns if p["candle"]["pattern"] != "regular"],
"multi_candle_patterns": multi_patterns,
"overall_bias": MultiCandleClassifier._aggregate_bias(single_patterns, multi_patterns),
}
@staticmethod
def _aggregate_bias(singles: list, multis: list) -> dict:
scores = []
for p in singles:
bias = p["candle"]["bias"]
strength = p["candle"]["strength"]
if "bullish" in bias: scores.append(strength)
elif "bearish" in bias: scores.append(-strength)
for p in multis:
if "bullish" in p["bias"]: scores.append(p["strength"])
elif "bearish" in p["bias"]: scores.append(-p["strength"])
avg = np.mean(scores) if scores else 0
return {"score": round(avg, 3), "direction": "BULLISH" if avg > 0.2 else "BEARISH" if avg < -0.2 else "NEUTRAL"}
Candlestick Statistics Engine
Candlestick Statistics Engine
CANDLE_STATS = {
"hammer": {"bullish_pct": 60, "avg_follow_through": 1.5, "sample_note": "Based on studies of 10+ years of FX data"},
"shooting_star": {"bearish_pct": 59, "avg_follow_through": 1.4},
"bullish_engulfing": {"bullish_pct": 63, "avg_follow_through": 1.8, "best_at": "support levels"},
"bearish_engulfing": {"bearish_pct": 62, "avg_follow_through": 1.7, "best_at": "resistance levels"},
"morning_star": {"bullish_pct": 68, "avg_follow_through": 2.2, "note": "High reliability when volume confirms"},
"evening_star": {"bearish_pct": 67, "avg_follow_through": 2.1},
"doji": {"reversal_pct": 51, "note": "Doji alone is weak — needs context (at key level + trend exhaustion)"},
"pin_bar": {"reversal_pct": 65, "avg_follow_through": 1.6, "best_at": "key levels with long wick into liquidity"},
"three_white_soldiers": {"bullish_pct": 72, "avg_follow_through": 2.5, "note": "Strongest multi-candle bullish pattern"},
"three_black_crows": {"bearish_pct": 71, "avg_follow_through": 2.4},
"marubozu": {"continuation_pct": 56, "note": "Shows conviction but often follows through only partially"},
"inside_bar": {"breakout_follow_pct": 62, "note": "Trade the breakout direction, not the inside bar itself"},
}
class CandleStatsEngine:
@staticmethod
def lookup(pattern: str) -> dict:
return CANDLE_STATS.get(pattern, {"error": f"No stats for {pattern}"})
@staticmethod
def all_stats() -> dict:
return CANDLE_STATS
Chart Pattern Recognition Vision
Chart Pattern Recognition Vision
Overview
Identifies classical chart patterns from images using price contour extraction, peak/trough detection on the extracted price curve, and geometric validation of pattern structures.
Stack
- OpenCV 4.13 — contour analysis, convex hull, template matching
- scikit-image 0.26 — ridge detection, peak finding on contours
- scipy 1.17 — peak finding, curve fitting, geometric analysis
1. Price Curve Extraction from Image
import cv2
import numpy as np
from scipy.signal import find_peaks, savgol_filter
from scipy.ndimage import gaussian_filter1d
class PriceCurveExtractor:
"""Extract the price curve as a 1D signal from a chart image."""
@staticmethod
def extract_price_curve(img: np.ndarray) -> np.ndarray:
"""
Extract price line from candlestick chart image.
Uses candle body midpoints as the price curve.
Returns: 1D array of y-positions (inverted: lower y = higher price).
"""
hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
h, w = img.shape[:2]
# Combine bullish + bearish candle masks
green = cv2.inRange(hsv, (35, 30, 30), (85, 255, 255))
red1 = cv2.inRange(hsv, (0, 30, 30), (15, 255, 255))
red2 = cv2.inRange(hsv, (165, 30, 30), (180, 255, 255))
candle_mask = cv2.bitwise_or(green, cv2.bitwise_or(red1, red2))
# Morphological cleanup
kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (3, 5))
candle_mask = cv2.morphologyEx(candle_mask, cv2.MORPH_CLOSE, kernel, iterations=2)
# For each column, find the center of candle pixels
price_curve = np.full(w, np.nan)
for col in range(w):
col_pixels = np.where(candle_mask[:, col] > 0)[0]
if len(col_pixels) > 2:
price_curve[col] = np.mean(col_pixels)
# Interpolate NaN gaps
valid = ~np.isnan(price_curve)
if np.sum(valid) < 20:
return np.array([])
from numpy import interp
x_valid = np.where(valid)[0]
price_curve = interp(np.arange(w), x_valid, price_curve[valid])
# Smooth
price_curve = savgol_filter(price_curve, window_length=min(21, len(price_curve) // 5 * 2 + 1),
polyorder=3)
return price_curve
@staticmethod
def find_swings(curve: np.ndarray, order: int = 15) -> dict:
"""Find swing highs and lows from the extracted price curve."""
# Invert because lower y = higher price in image coordinates
inv_curve = -curve
peaks, peak_props = find_peaks(inv_curve, distance=order, prominence=np.std(curve) * 0.3)
troughs, trough_props = find_peaks(curve, distance=order, prominence=np.std(curve) * 0.3)
return {
"highs": [{"x": int(p), "y": int(curve[p])} for p in peaks],
"lows": [{"x": int(t), "y": int(curve[t])} for t in troughs],
"n_highs": len(peaks),
"n_lows": len(troughs),
}
2. Pattern Detectors (Image-Based)
class VisualPatternDetector:
"""Detect chart patterns from extracted price curve and swing points."""
@staticmethod
def detect_head_and_shoulders(swings: dict, curve: np.ndarray, img_height: int) -> list[dict]:
"""H&S from image: three swing highs where middle is highest."""
patterns = []
highs = swings["highs"]
if len(highs) < 3:
return patterns
for i in range(len(highs) - 2):
ls, head, rs = highs[i], highs[i+1], highs[i+2]
# Head must be highest (lowest y value = highest price)
if head["y"] < ls["y"] and head["y"] < rs["y"]:
shoulder_diff = abs(ls["y"] - rs["y"]) / img_height
if shoulder_diff < 0.05: # Shoulders at similar level
neckline_y = max(ls["y"], rs["y"]) # Higher y = lower price
height = neckline_y - head["y"]
patterns.append({
"pattern": "head_and_shoulders",
"head": head, "left_shoulder": ls, "right_shoulder": rs,
"neckline_y": neckline_y,
"height_px": height,
"target_y": neckline_y + height,
"bias": "BEARISH",
"reliability": 0.83,
"confidence": round(1 - shoulder_diff / 0.05, 2),
})
# Inverse H&S (from lows)
lows = swings["lows"]
if len(lows) >= 3:
for i in range(len(lows) - 2):
ls, head, rs = lows[i], lows[i+1], lows[i+2]
if head["y"] > ls["y"] and head["y"] > rs["y"]:
shoulder_diff = abs(ls["y"] - rs["y"]) / img_height
if shoulder_diff < 0.05:
neckline_y = min(ls["y"], rs["y"])
height = head["y"] - neckline_y
patterns.append({
"pattern": "inverse_head_and_shoulders",
"head": head, "left_shoulder": ls, "right_shoulder": rs,
"neckline_y": neckline_y,
"height_px": height,
"bias": "BULLISH",
"reliability": 0.83,
})
return patterns
@staticmethod
def detect_double_top_bottom(swings: dict, img_height: int, tolerance: float = 0.02) -> list[dict]:
patterns = []
highs = swings["highs"]
lows = swings["lows"]
for i in range(len(highs) - 1):
h1, h2 = highs[i], highs[i+1]
diff = abs(h1["y"] - h2["y"]) / img_height
if diff < tolerance:
patterns.append({"pattern": "double_top", "peak1": h1, "peak2": h2,
"bias": "BEARISH", "reliability": 0.72})
for i in range(len(lows) - 1):
l1, l2 = lows[i], lows[i+1]
diff = abs(l1["y"] - l2["y"]) / img_height
if diff < tolerance:
patterns.append({"pattern": "double_bottom", "trough1": l1, "trough2": l2,
"bias": "BULLISH", "reliability": 0.72})
return patterns
@staticmethod
def detect_triangles(swings: dict, curve: np.ndarray) -> list[dict]:
"""Detect converging trendlines forming triangles."""
patterns = []
highs = swings["highs"]
lows = swings["lows"]
if len(highs) >= 2 and len(lows) >= 2:
h_slope = (highs[-1]["y"] - highs[0]["y"]) / max(highs[-1]["x"] - highs[0]["x"], 1)
l_slope = (lows[-1]["y"] - lows[0]["y"]) / max(lows[-1]["x"] - lows[0]["x"], 1)
# Remember: positive slope in image = price declining
if abs(h_slope) < 0.05 and l_slope < -0.05:
patterns.append({"pattern": "ascending_triangle", "bias": "BULLISH", "reliability": 0.73,
"h_slope": round(h_slope, 4), "l_slope": round(l_slope, 4)})
elif abs(l_slope) < 0.05 and h_slope > 0.05:
patterns.append({"pattern": "descending_triangle", "bias": "BEARISH", "reliability": 0.72,
"h_slope": round(h_slope, 4), "l_slope": round(l_slope, 4)})
elif h_slope > 0.02 and l_slope < -0.02:
patterns.append({"pattern": "symmetrical_triangle", "bias": "NEUTRAL", "reliability": 0.60})
elif h_slope < -0.02 and l_slope < -0.02:
patterns.append({"pattern": "falling_wedge", "bias": "BULLISH", "reliability": 0.68})
elif h_slope > 0.02 and l_slope > 0.02:
patterns.append({"pattern": "rising_wedge", "bias": "BEARISH", "reliability": 0.68})
return patterns
@staticmethod
def full_pattern_scan(img: np.ndarray) -> dict:
"""Complete visual pattern recognition pipeline."""
curve = PriceCurveExtractor.extract_price_curve(img)
if len(curve) == 0:
return {"error": "Could not extract price curve from image"}
h, w = img.shape[:2]
swings = PriceCurveExtractor.find_swings(curve)
hs_patterns = VisualPatternDetector.detect_head_and_shoulders(swings, curve, h)
dt_patterns = VisualPatternDetector.detect_double_top_bottom(swings, h)
tri_patterns = VisualPatternDetector.detect_triangles(swings, curve)
all_patterns = hs_patterns + dt_patterns + tri_patterns
return {
"swings": swings,
"patterns_found": len(all_patterns),
"patterns": sorted(all_patterns, key=lambda p: p.get("reliability", 0), reverse=True),
"dominant_pattern": all_patterns[0] if all_patterns else None,
"overall_bias": VisualPatternDetector._aggregate_pattern_bias(all_patterns),
}
@staticmethod
def _aggregate_pattern_bias(patterns: list) -> str:
if not patterns: return "NO PATTERNS"
bullish = sum(1 for p in patterns if "BULLISH" in p.get("bias", ""))
bearish = sum(1 for p in patterns if "BEARISH" in p.get("bias", ""))
if bullish > bearish: return "BULLISH"
if bearish > bullish: return "BEARISH"
return "NEUTRAL"
Harmonic Pattern Engine
Harmonic Pattern Engine
import pandas as pd, numpy as np
from scipy.signal import argrelextrema
HARMONIC_RATIOS = {
"gartley": {"XB": (0.618, 0.618), "AC": (0.382, 0.886), "BD": (1.272, 1.618), "XD": (0.786, 0.786)},
"butterfly": {"XB": (0.786, 0.786), "AC": (0.382, 0.886), "BD": (1.618, 2.618), "XD": (1.272, 1.618)},
"bat": {"XB": (0.382, 0.500), "AC": (0.382, 0.886), "BD": (1.618, 2.618), "XD": (0.886, 0.886)},
"crab": {"XB": (0.382, 0.618), "AC": (0.382, 0.886), "BD": (2.240, 3.618), "XD": (1.618, 1.618)},
"cypher": {"XB": (0.382, 0.618), "AC": (1.130, 1.414), "BD": (1.272, 2.000), "XD": (0.786, 0.786)},
}
class HarmonicEngine:
@staticmethod
def detect_xabcd(df: pd.DataFrame, tolerance: float = 0.05) -> list[dict]:
"""Detect XABCD harmonic patterns from swing points."""
highs = argrelextrema(df["high"].values, np.greater, order=5)[0]
lows = argrelextrema(df["low"].values, np.less, order=5)[0]
swings = []
for i in highs: swings.append({"idx": i, "price": df["high"].iloc[i], "type": "H"})
for i in lows: swings.append({"idx": i, "price": df["low"].iloc[i], "type": "L"})
swings.sort(key=lambda s: s["idx"])
patterns = []
for i in range(len(swings) - 4):
X, A, B, C, D = [swings[j]["price"] for j in range(i, i + 5)]
XA = abs(A - X)
if XA == 0: continue
AB = abs(B - A)
BC = abs(C - B)
CD = abs(D - C)
XB_ratio = AB / XA
AC_ratio = BC / AB if AB > 0 else 0
BD_ratio = CD / BC if BC > 0 else 0
XD_ratio = abs(D - X) / XA
for name, ratios in HARMONIC_RATIOS.items():
xb_min, xb_max = ratios["XB"][0] - tolerance, ratios["XB"][1] + tolerance
xd_min, xd_max = ratios["XD"][0] - tolerance, ratios["XD"][1] + tolerance
if xb_min <= XB_ratio <= xb_max and xd_min <= XD_ratio <= xd_max:
bullish = D < X if swings[i]["type"] == "L" else D > X
patterns.append({
"pattern": name,
"bullish": bullish,
"X": round(X, 5), "A": round(A, 5), "B": round(B, 5), "C": round(C, 5), "D": round(D, 5),
"XB": round(XB_ratio, 3), "XD": round(XD_ratio, 3),
"prz": round(D, 5),
"signal": f"{'BUY' if bullish else 'SELL'} at PRZ {round(D, 5)}",
"stop": round(X, 5),
"tp1": round(D + (A - D) * 0.382, 5) if bullish else round(D - (D - A) * 0.382, 5),
"tp2": round(D + (A - D) * 0.618, 5) if bullish else round(D - (D - A) * 0.618, 5),
})
return patterns
Elliott Wave Engine
Elliott Wave Engine
import pandas as pd, numpy as np
from scipy.signal import argrelextrema
class ElliottWaveEngine:
@staticmethod
def find_waves(df: pd.DataFrame, order: int = 10) -> dict:
"""Attempt to identify Elliott Wave structure from swing points."""
highs_idx = argrelextrema(df["high"].values, np.greater, order=order)[0]
lows_idx = argrelextrema(df["low"].values, np.less, order=order)[0]
swings = []
for i in highs_idx:
swings.append({"idx": int(i), "price": df["high"].iloc[i], "type": "high", "time": df.index[i]})
for i in lows_idx:
swings.append({"idx": int(i), "price": df["low"].iloc[i], "type": "low", "time": df.index[i]})
swings.sort(key=lambda s: s["idx"])
# Validate impulse wave rules
waves = ElliottWaveEngine._classify_impulse(swings)
return {
"swings_found": len(swings),
"waves": waves,
"current_wave": waves[-1] if waves else None,
"note": "Elliott Waves are subjective. Multiple valid counts often exist. Use as confluence, not primary signal.",
}
@staticmethod
def _classify_impulse(swings: list) -> list:
"""Check if swing sequence follows 5-wave impulse rules."""
waves = []
if len(swings) < 5:
return [{"wave": "insufficient_data", "swings": len(swings)}]
for i in range(0, len(swings) - 4, 2):
s = swings[i:i+5]
if len(s) < 5: break
# Basic impulse: up-down-up-down-up (bullish) or reverse
is_bullish = s[0]["type"] == "low" and s[2]["price"] > s[0]["price"]
if is_bullish:
w3_longest = (s[2]["price"] - s[1]["price"]) > (s[0]["price"] if s[0]["type"]=="high" else 0)
w2_above_w1_start = s[1]["price"] > s[0]["price"]
waves.append({
"type": "impulse_bullish",
"wave_1": {"start": round(s[0]["price"], 5), "end": round(s[1]["price"], 5)},
"wave_2": {"start": round(s[1]["price"], 5), "end": round(s[2]["price"], 5) if len(s) > 2 else 0},
"w2_valid": w2_above_w1_start,
"position": i,
})
return waves if waves else [{"wave": "no_clear_impulse"}]
@staticmethod
def fibonacci_targets(wave_1_start: float, wave_1_end: float, wave_2_end: float) -> dict:
"""Project wave 3 and wave 5 targets using Fibonacci extensions."""
w1_range = abs(wave_1_end - wave_1_start)
direction = 1 if wave_1_end > wave_1_start else -1
return {
"wave_3_targets": {
"1.000": round(wave_2_end + direction * w1_range * 1.0, 5),
"1.618": round(wave_2_end + direction * w1_range * 1.618, 5),
"2.618": round(wave_2_end + direction * w1_range * 2.618, 5),
},
"wave_5_note": "Project from wave 4 end using wave 1 range",
"invalidation": round(wave_1_start, 5),
}
Trendline Sr Vision
Trendline & Support/Resistance Vision
Overview
Computer vision algorithms for detecting trendlines and S/R levels directly from chart images. Uses Hough line transforms, Line Segment Detector (LSD), horizontal density projection, and price level clustering. No price data needed — pure image analysis.
Stack
- OpenCV 4.13 — HoughLinesP, LSD (ximgproc), edge detection
- scikit-image 0.26 — probabilistic Hough, peak detection
- scipy 1.17 — signal peak finding, clustering
1. Support/Resistance Level Detection
import cv2
import numpy as np
from scipy.signal import find_peaks
from scipy.ndimage import gaussian_filter1d
from scipy.cluster.hierarchy import fcluster, linkage
from skimage.transform import probabilistic_hough_line
from skimage import feature
class SRDetector:
"""Detect support and resistance levels from chart images."""
@staticmethod
def horizontal_projection_sr(img: np.ndarray, n_levels: int = 8) -> list[dict]:
"""
Project pixel intensity horizontally to find price levels where
price action clusters (many candle bodies/wicks at same level).
"""
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
h, w = gray.shape
# Invert if dark theme (candles are bright on dark background)
if np.mean(gray) < 128:
gray = 255 - gray
# Horizontal projection: sum intensity per row
projection = np.sum(gray.astype(float), axis=1)
# Smooth to reduce noise
projection = gaussian_filter1d(projection, sigma=h * 0.01)
# Find peaks (rows with high pixel density = S/R levels)
peaks, properties = find_peaks(
projection,
height=np.percentile(projection, 75),
distance=h // (n_levels * 2),
prominence=np.std(projection) * 0.5,
)
# Cluster nearby peaks
if len(peaks) > n_levels:
peaks = peaks[np.argsort(properties["peak_heights"])[-n_levels:]]
levels = []
for peak in sorted(peaks):
y_pct = 1 - peak / h # Invert: top of image = high price
strength = projection[peak] / projection.max()
levels.append({
"y_pixel": int(peak),
"price_pct": round(y_pct, 4), # 0=bottom, 1=top of chart
"strength": round(strength, 3),
"type": "resistance" if y_pct > 0.5 else "support",
})
return sorted(levels, key=lambda l: l["strength"], reverse=True)
@staticmethod
def candle_body_clustering_sr(candles: list, n_clusters: int = 6) -> list[dict]:
"""
Cluster candle body edges (open/close levels) to find natural S/R zones.
Uses hierarchical clustering on candle geometry from CandleExtractor.
"""
if not candles or len(candles) < 5:
return []
# Collect all body edges (top and bottom of each candle body)
edges = []
for c in candles:
edges.extend([c.body_top, c.body_bottom, c.y_top, c.y_bottom])
edges = np.array(edges).reshape(-1, 1)
# Hierarchical clustering
Z = linkage(edges, method="ward")
labels = fcluster(Z, t=n_clusters, criterion="maxclust")
# Compute cluster centers and sizes
levels = []
for cluster_id in range(1, n_clusters + 1):
cluster_edges = edges[labels == cluster_id]
if len(cluster_edges) < 3:
continue
center = np.mean(cluster_edges)
spread = np.std(cluster_edges)
levels.append({
"y_pixel": int(center),
"zone_width_px": int(spread * 2),
"touch_count": len(cluster_edges),
"strength": round(len(cluster_edges) / len(edges), 3),
})
return sorted(levels, key=lambda l: l["strength"], reverse=True)
2. Trendline Detection
class TrendlineDetector:
"""Detect trendlines from chart images using line detection algorithms."""
@staticmethod
def hough_trendlines(img: np.ndarray, min_length: int = 100, max_gap: int = 10) -> list[dict]:
"""
Probabilistic Hough Line Transform for trendline detection.
OpenCV 4.13 HoughLinesP.
"""
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
edges = cv2.Canny(gray, 50, 150, apertureSize=3)
# Dilate edges slightly to connect fragmented lines
edges = cv2.dilate(edges, np.ones((2, 2), np.uint8), iterations=1)
lines = cv2.HoughLinesP(edges, rho=1, theta=np.pi / 180, threshold=80,
minLineLength=min_length, maxLineGap=max_gap)
if lines is None:
return []
trendlines = []
h, w = gray.shape
for line in lines:
x1, y1, x2, y2 = line[0]
angle = np.degrees(np.arctan2(y2 - y1, x2 - x1))
length = np.sqrt((x2 - x1) ** 2 + (y2 - y1) ** 2)
# Filter: trendlines should be roughly diagonal (10-80 degrees)
# or horizontal (S/R lines, 0-10 degrees)
abs_angle = abs(angle)
if abs_angle > 85: # Vertical lines = not trendlines
continue
line_type = TrendlineDetector._classify_line(angle, y1, y2, h)
trendlines.append({
"start": (x1, y1), "end": (x2, y2),
"angle": round(angle, 1),
"length": round(length, 1),
"type": line_type,
"slope": round((y2 - y1) / max(x2 - x1, 1), 4),
})
# Merge nearby parallel lines
trendlines = TrendlineDetector._m
…(truncated)