"""Fractal pattern detection and encoding for candlestick sequences.""" from __future__ import annotations from dataclasses import dataclass from enum import IntEnum import numpy as np import pandas as pd class CandleType(IntEnum): DOWN = -1 DOJI = 0 UP = 1 @dataclass class FractalPattern: sequence: tuple[CandleType, ...] start_idx: int end_idx: int body_ratios: tuple[float, ...] shadow_ratios: tuple[float, ...] @property def length(self) -> int: return len(self.sequence) @property def code(self) -> str: m = {CandleType.DOWN: "D", CandleType.DOJI: "0", CandleType.UP: "U"} return "".join(m[c] for c in self.sequence) def __hash__(self) -> int: return hash(self.sequence) def classify_candle(open_: float, close: float, high: float, low: float, doji_threshold: float = 0.1) -> CandleType: body = abs(close - open_) total_range = high - low if total_range == 0: return CandleType.DOJI ratio = body / total_range if ratio < doji_threshold: return CandleType.DOJI return CandleType.UP if close > open_ else CandleType.DOWN def encode_candles(df: pd.DataFrame, doji_threshold: float = 0.1) -> np.ndarray: types = np.zeros(len(df), dtype=np.int8) body = np.abs(df["close"].values - df["open"].values) total = df["high"].values - df["low"].values total = np.where(total == 0, 1e-10, total) ratio = body / total types[ratio < doji_threshold] = CandleType.DOJI bullish = (ratio >= doji_threshold) & (df["close"].values > df["open"].values) bearish = (ratio >= doji_threshold) & (df["close"].values <= df["open"].values) types[bullish] = CandleType.UP types[bearish] = CandleType.DOWN return types def extract_body_ratios(df: pd.DataFrame) -> np.ndarray: body = np.abs(df["close"].values - df["open"].values) total = df["high"].values - df["low"].values total = np.where(total == 0, 1e-10, total) return body / total def extract_shadow_ratios(df: pd.DataFrame) -> np.ndarray: o, c, h, l = df["open"].values, df["close"].values, df["high"].values, df["low"].values upper_shadow = h - np.maximum(o, c) lower_shadow = np.minimum(o, c) - l total = h - l total = np.where(total == 0, 1e-10, total) return (upper_shadow - lower_shadow) / total def find_patterns(df: pd.DataFrame, min_len: int = 3, max_len: int = 6, doji_threshold: float = 0.1) -> list[FractalPattern]: candle_types = encode_candles(df, doji_threshold) body_ratios = extract_body_ratios(df) shadow_ratios = extract_shadow_ratios(df) patterns: list[FractalPattern] = [] for length in range(min_len, max_len + 1): for i in range(len(df) - length): seq = tuple(CandleType(t) for t in candle_types[i : i + length]) br = tuple(body_ratios[i : i + length]) sr = tuple(shadow_ratios[i : i + length]) patterns.append(FractalPattern( sequence=seq, start_idx=i, end_idx=i + length, body_ratios=br, shadow_ratios=sr, )) return patterns def pattern_frequency(patterns: list[FractalPattern]) -> pd.DataFrame: from collections import Counter codes = [p.code for p in patterns] counts = Counter(codes) total = len(codes) rows = [ {"pattern": code, "count": cnt, "freq": cnt / total, "length": len(code)} for code, cnt in counts.most_common() ] return pd.DataFrame(rows) def normalize_pattern_window(df: pd.DataFrame, start: int, end: int) -> np.ndarray: """Normalize OHLC window to [0,1] range for comparison.""" window = df.iloc[start:end][["open", "high", "low", "close"]].values mn = window.min() mx = window.max() if mx - mn == 0: return np.zeros_like(window) return (window - mn) / (mx - mn)