Krikorios
48e60d015f
- Add advanced metrics dashboard with trade analytics - Add new trading components (EntryTypeAnalysis, MultiDayPositionTracker, NewsEventTracker, etc.) - Add strategy mode selector and trend confirmation - Add risk automation panel and slippage correlation analysis - Add daily trading plan enhancements with modal components - Add custom hooks (useApi, useLocalStorage, useAdvancedTradeMetrics) - Add broker service integration and trading API - Add test setup and vitest configuration - Include parquet data files for live market data - Add comprehensive documentation in docs/ folder
388 lines
15 KiB
Python
388 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Iterable, List, Sequence, Tuple, Union
|
|
|
|
from app.schemas.schemas import PatternSignal, PriceData
|
|
|
|
BarLike = Union[PriceData, dict]
|
|
|
|
|
|
@dataclass
|
|
class Candle:
|
|
time: int
|
|
open: float
|
|
high: float
|
|
low: float
|
|
close: float
|
|
|
|
@property
|
|
def hl2(self) -> float:
|
|
return (self.high + self.low) / 2
|
|
|
|
|
|
class CandlestickPatternDetector:
|
|
"""Translated subset of the TradingView *All Candlestick Patterns* study.
|
|
|
|
The detector focuses on high-signal patterns that are most useful for
|
|
risk automation and narrative building. The implementation is intentionally
|
|
modular so additional patterns from the Pine script can be ported quickly.
|
|
"""
|
|
|
|
BODY_AVG_EMA = 14
|
|
SHADOW_PERCENT = 5.0
|
|
SHADOW_EQUALS_PERCENT = 100.0
|
|
DOJI_BODY_PERCENT = 5.0
|
|
LONG_LOWER_PERCENT = 75.0
|
|
LONG_UPPER_PERCENT = 75.0
|
|
HAMMER_FACTOR = 2.0
|
|
TREND_SMA = 50
|
|
TREND_SMA_LONG = 200
|
|
|
|
def analyze(self, rows: Iterable[BarLike]) -> List[PatternSignal]:
|
|
candles = self._normalize(rows)
|
|
if len(candles) < 3:
|
|
return []
|
|
|
|
opens = [c.open for c in candles]
|
|
highs = [c.high for c in candles]
|
|
lows = [c.low for c in candles]
|
|
closes = [c.close for c in candles]
|
|
times = [c.time for c in candles]
|
|
|
|
body_hi = [max(o, c) for o, c in zip(opens, closes)]
|
|
body_lo = [min(o, c) for o, c in zip(opens, closes)]
|
|
bodies = [hi - lo for hi, lo in zip(body_hi, body_lo)]
|
|
ranges = [h - l for h, l in zip(highs, lows)]
|
|
upper_shadows = [h - hi for h, hi in zip(highs, body_hi)]
|
|
lower_shadows = [lo - l for lo, l in zip(body_lo, lows)]
|
|
body_avg = self._ema_series(bodies, self.BODY_AVG_EMA)
|
|
sma50 = self._sma_series(closes, self.TREND_SMA)
|
|
sma200 = self._sma_series(closes, self.TREND_SMA_LONG)
|
|
|
|
up_trend = [False] * len(candles)
|
|
down_trend = [False] * len(candles)
|
|
for idx in range(len(candles)):
|
|
if sma50[idx] is None:
|
|
if idx > 0:
|
|
up_trend[idx] = closes[idx] > closes[idx - 1]
|
|
down_trend[idx] = closes[idx] < closes[idx - 1]
|
|
continue
|
|
close = closes[idx]
|
|
s50 = sma50[idx]
|
|
s200 = sma200[idx]
|
|
up = close > s50
|
|
down = close < s50
|
|
if s200 is not None:
|
|
up = up and s50 > s200
|
|
down = down and s50 < s200
|
|
up_trend[idx] = up
|
|
down_trend[idx] = down
|
|
|
|
pattern_signals: List[PatternSignal] = []
|
|
|
|
for i in range(len(candles)):
|
|
detected = self._detect_at(
|
|
i,
|
|
candles,
|
|
body_hi,
|
|
body_lo,
|
|
bodies,
|
|
body_avg,
|
|
ranges,
|
|
upper_shadows,
|
|
lower_shadows,
|
|
up_trend,
|
|
down_trend,
|
|
)
|
|
for pattern, classification in detected:
|
|
pattern_signals.append(
|
|
PatternSignal(
|
|
pattern=pattern,
|
|
classification=classification,
|
|
price=closes[i],
|
|
time=times[i],
|
|
)
|
|
)
|
|
|
|
return pattern_signals
|
|
|
|
# ------------------------------------------------------------------
|
|
# Detection helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
def _detect_at(
|
|
self,
|
|
i: int,
|
|
candles: Sequence[Candle],
|
|
body_hi: Sequence[float],
|
|
body_lo: Sequence[float],
|
|
bodies: Sequence[float],
|
|
body_avg: Sequence[float | None],
|
|
ranges: Sequence[float],
|
|
upper_shadows: Sequence[float],
|
|
lower_shadows: Sequence[float],
|
|
up_trend: Sequence[bool],
|
|
down_trend: Sequence[bool],
|
|
) -> List[Tuple[str, str]]:
|
|
signals: List[Tuple[str, str]] = []
|
|
|
|
if i == 0:
|
|
return signals
|
|
|
|
body = bodies[i]
|
|
body_average = body_avg[i] or 0.0
|
|
range_ = ranges[i]
|
|
upper = upper_shadows[i]
|
|
lower = lower_shadows[i]
|
|
is_white = candles[i].close > candles[i].open
|
|
is_black = candles[i].open > candles[i].close
|
|
prev_white = candles[i - 1].close > candles[i - 1].open
|
|
prev_black = candles[i - 1].open > candles[i - 1].close
|
|
small_body = body_average > 0 and body < body_average
|
|
long_body = body_average > 0 and body > body_average
|
|
has_upper_shadow = upper > self.SHADOW_PERCENT / 100 * body if body > 0 else False
|
|
has_lower_shadow = lower > self.SHADOW_PERCENT / 100 * body if body > 0 else False
|
|
doji = self._is_doji(body, range_)
|
|
|
|
# Single-candle patterns -------------------------------------------------
|
|
if doji:
|
|
signals.append(("Doji", "NEUTRAL"))
|
|
if upper <= body:
|
|
signals.append(("Dragonfly Doji", "BULLISH"))
|
|
if lower <= body:
|
|
signals.append(("Gravestone Doji", "BEARISH"))
|
|
|
|
if body > 0:
|
|
if not has_upper_shadow and lower >= self.HAMMER_FACTOR * body and candles[i].hl2 < body_lo[i] and down_trend[i]:
|
|
signals.append(("Hammer", "BULLISH"))
|
|
if not has_upper_shadow and lower >= self.HAMMER_FACTOR * body and candles[i].hl2 < body_lo[i] and up_trend[i]:
|
|
signals.append(("Hanging Man", "BEARISH"))
|
|
if not has_lower_shadow and upper >= self.HAMMER_FACTOR * body and candles[i].hl2 > body_hi[i] and down_trend[i]:
|
|
signals.append(("Inverted Hammer", "BULLISH"))
|
|
if not has_lower_shadow and upper >= self.HAMMER_FACTOR * body and candles[i].hl2 > body_hi[i] and up_trend[i]:
|
|
signals.append(("Shooting Star", "BEARISH"))
|
|
|
|
if body > 0 and upper <= body * self.SHADOW_PERCENT / 100 and lower <= body * self.SHADOW_PERCENT / 100:
|
|
if is_white:
|
|
signals.append(("Marubozu White", "BULLISH"))
|
|
if is_black:
|
|
signals.append(("Marubozu Black", "BEARISH"))
|
|
|
|
if lower > range_ * self.LONG_LOWER_PERCENT / 100:
|
|
signals.append(("Long Lower Shadow", "BULLISH"))
|
|
if upper > range_ * self.LONG_UPPER_PERCENT / 100:
|
|
signals.append(("Long Upper Shadow", "BEARISH"))
|
|
|
|
# Multi-candle patterns --------------------------------------------------
|
|
signals.extend(
|
|
self._two_candle_patterns(
|
|
i,
|
|
candles,
|
|
body_hi,
|
|
body_lo,
|
|
bodies,
|
|
body_avg,
|
|
ranges,
|
|
up_trend,
|
|
down_trend,
|
|
)
|
|
)
|
|
signals.extend(
|
|
self._three_candle_patterns(
|
|
i,
|
|
candles,
|
|
body_hi,
|
|
body_lo,
|
|
bodies,
|
|
body_avg,
|
|
up_trend,
|
|
down_trend,
|
|
)
|
|
)
|
|
signals.extend(self._soldiers_and_crows(i, candles, bodies, body_avg))
|
|
|
|
return signals
|
|
|
|
def _two_candle_patterns(
|
|
self,
|
|
i: int,
|
|
candles: Sequence[Candle],
|
|
body_hi: Sequence[float],
|
|
body_lo: Sequence[float],
|
|
bodies: Sequence[float],
|
|
body_avg: Sequence[float | None],
|
|
ranges: Sequence[float],
|
|
up_trend: Sequence[bool],
|
|
down_trend: Sequence[bool],
|
|
) -> List[Tuple[str, str]]:
|
|
if i < 1:
|
|
return []
|
|
signals: List[Tuple[str, str]] = []
|
|
body = bodies[i]
|
|
body_prev = bodies[i - 1]
|
|
avg = body_avg[i] or 0.0
|
|
avg_prev = body_avg[i - 1] or 0.0
|
|
white = candles[i].close > candles[i].open
|
|
black = candles[i].open > candles[i].close
|
|
prev_white = candles[i - 1].close > candles[i - 1].open
|
|
prev_black = candles[i - 1].open > candles[i - 1].close
|
|
|
|
tol = (avg + avg_prev) / 2 * 0.05 if (avg + avg_prev) > 0 else 0.0
|
|
|
|
# Tweezer patterns
|
|
if abs(candles[i].high - candles[i - 1].high) <= tol and prev_white and black and up_trend[i - 1]:
|
|
signals.append(("Tweezer Top", "BEARISH"))
|
|
if abs(candles[i].low - candles[i - 1].low) <= tol and prev_black and white and down_trend[i - 1]:
|
|
signals.append(("Tweezer Bottom", "BULLISH"))
|
|
|
|
# Engulfing
|
|
if down_trend[i - 1] and prev_black and (avg_prev == 0 or body_prev <= avg_prev) and white:
|
|
if candles[i].close >= candles[i - 1].open and candles[i].open <= candles[i - 1].close:
|
|
signals.append(("Bullish Engulfing", "BULLISH"))
|
|
if up_trend[i - 1] and prev_white and (avg_prev == 0 or body_prev <= avg_prev) and black:
|
|
if candles[i].close <= candles[i - 1].open and candles[i].open >= candles[i - 1].close:
|
|
signals.append(("Bearish Engulfing", "BEARISH"))
|
|
|
|
# Piercing / Dark Cloud Cover
|
|
mid_prev = (candles[i - 1].open + candles[i - 1].close) / 2
|
|
if down_trend[i - 1] and prev_black and white:
|
|
if candles[i].open <= candles[i - 1].low and candles[i].close > mid_prev and candles[i].close < candles[i - 1].open:
|
|
signals.append(("Piercing", "BULLISH"))
|
|
if up_trend[i - 1] and prev_white and black:
|
|
if candles[i].open >= candles[i - 1].high and candles[i].close < mid_prev and candles[i].close > candles[i - 1].open:
|
|
signals.append(("Dark Cloud Cover", "BEARISH"))
|
|
|
|
# Doji Star variants
|
|
if self._is_doji(body, ranges[i]) and up_trend[i - 1] and prev_white:
|
|
if candles[i].open > candles[i - 1].high:
|
|
signals.append(("Doji Star", "BEARISH"))
|
|
if self._is_doji(body, ranges[i]) and down_trend[i - 1] and prev_black:
|
|
if candles[i].open < candles[i - 1].low:
|
|
signals.append(("Doji Star", "BULLISH"))
|
|
|
|
return signals
|
|
|
|
def _three_candle_patterns(
|
|
self,
|
|
i: int,
|
|
candles: Sequence[Candle],
|
|
body_hi: Sequence[float],
|
|
body_lo: Sequence[float],
|
|
bodies: Sequence[float],
|
|
body_avg: Sequence[float | None],
|
|
up_trend: Sequence[bool],
|
|
down_trend: Sequence[bool],
|
|
) -> List[Tuple[str, str]]:
|
|
if i < 2:
|
|
return []
|
|
signals: List[Tuple[str, str]] = []
|
|
|
|
c0, c1, c2 = candles[i - 2], candles[i - 1], candles[i]
|
|
body0, body1, body2 = bodies[i - 2], bodies[i - 1], bodies[i]
|
|
avg0 = body_avg[i - 2] or 0.0
|
|
avg1 = body_avg[i - 1] or 0.0
|
|
avg2 = body_avg[i] or 0.0
|
|
white2 = c2.close > c2.open
|
|
black2 = c2.open > c2.close
|
|
small1 = avg1 > 0 and body1 < avg1
|
|
doji1 = self._is_doji(body1, c1.high - c1.low)
|
|
|
|
mid0 = (c0.open + c0.close) / 2
|
|
|
|
if down_trend[i - 2] and (c0.open > c0.close) and small1 and white2:
|
|
if c1.open < c0.low and c2.close >= mid0 and c2.close < c0.high:
|
|
signals.append(("Morning Star", "BULLISH"))
|
|
if up_trend[i - 2] and (c0.close > c0.open) and small1 and black2:
|
|
if c1.open > c0.high and c2.close <= mid0 and c2.close > c0.low:
|
|
signals.append(("Evening Star", "BEARISH"))
|
|
|
|
if down_trend[i - 2] and (c0.open > c0.close) and doji1 and white2:
|
|
if c1.open < c0.low and c2.close >= mid0 and c2.close < c0.high:
|
|
signals.append(("Morning Doji Star", "BULLISH"))
|
|
if up_trend[i - 2] and (c0.close > c0.open) and doji1 and black2:
|
|
if c1.open > c0.high and c2.close <= mid0 and c2.close > c0.low:
|
|
signals.append(("Evening Doji Star", "BEARISH"))
|
|
|
|
return signals
|
|
|
|
def _soldiers_and_crows(
|
|
self,
|
|
i: int,
|
|
candles: Sequence[Candle],
|
|
bodies: Sequence[float],
|
|
body_avg: Sequence[float | None],
|
|
) -> List[Tuple[str, str]]:
|
|
if i < 2:
|
|
return []
|
|
signals: List[Tuple[str, str]] = []
|
|
c0, c1, c2 = candles[i - 2], candles[i - 1], candles[i]
|
|
body0, body1, body2 = bodies[i - 2], bodies[i - 1], bodies[i]
|
|
avg0 = body_avg[i - 2] or 0.0
|
|
avg1 = body_avg[i - 1] or 0.0
|
|
avg2 = body_avg[i] or 0.0
|
|
|
|
if all(b > a for b, a in zip((body0, body1, body2), (avg0, avg1, avg2))):
|
|
if c0.close < c0.open and c1.close > c1.open and c2.close > c2.open:
|
|
if c1.open > c0.close and c2.open > c1.close and c2.close > c1.close > c0.close:
|
|
signals.append(("Three White Soldiers", "BULLISH"))
|
|
if c0.close > c0.open and c1.close < c1.open and c2.close < c2.open:
|
|
if c1.open < c0.close and c2.open < c1.close and c2.close < c1.close < c0.close:
|
|
signals.append(("Three Black Crows", "BEARISH"))
|
|
|
|
return signals
|
|
|
|
# ------------------------------------------------------------------
|
|
# Utility functions
|
|
# ------------------------------------------------------------------
|
|
|
|
def _normalize(self, rows: Iterable[BarLike]) -> List[Candle]:
|
|
candles: List[Candle] = []
|
|
for row in rows:
|
|
if isinstance(row, PriceData):
|
|
candles.append(Candle(time=row.time, open=row.open, high=row.high, low=row.low, close=row.close))
|
|
else:
|
|
candles.append(
|
|
Candle(
|
|
time=int(row.get("time", len(candles))),
|
|
open=float(row["open"]),
|
|
high=float(row["high"]),
|
|
low=float(row["low"]),
|
|
close=float(row["close"]),
|
|
)
|
|
)
|
|
return candles
|
|
|
|
def _ema_series(self, values: Sequence[float], length: int) -> List[float | None]:
|
|
ema_series: List[float | None] = [None] * len(values)
|
|
if len(values) < length:
|
|
return ema_series
|
|
k = 2 / (length + 1)
|
|
ema = sum(values[:length]) / length
|
|
ema_series[length - 1] = ema
|
|
for idx in range(length, len(values)):
|
|
ema = values[idx] * k + ema * (1 - k)
|
|
ema_series[idx] = ema
|
|
return ema_series
|
|
|
|
def _sma_series(self, values: Sequence[float], length: int) -> List[float | None]:
|
|
sma_series: List[float | None] = [None] * len(values)
|
|
if length <= 0:
|
|
return sma_series
|
|
window_sum = 0.0
|
|
for idx, value in enumerate(values):
|
|
window_sum += value
|
|
if idx >= length:
|
|
window_sum -= values[idx - length]
|
|
if idx >= length - 1:
|
|
sma_series[idx] = window_sum / length
|
|
return sma_series
|
|
|
|
def _is_doji(self, body: float, candle_range: float) -> bool:
|
|
if candle_range <= 0:
|
|
return False
|
|
return body <= candle_range * self.DOJI_BODY_PERCENT / 100
|
|
|
|
|
|
candlestick_detector = CandlestickPatternDetector()
|