technical-price-action-patterns"Provides Analysis of candlestick and chart patterns for price movement prediction"
Independent GitHub snapshot · Automated classification · Original documentation governs
name: technical-price-action-patterns compatibility: opencode completeness: 95 content-types:
Role: Identify high-probability price patterns to forecast market direction
Philosophy: Price action reflects all market participants' collective sentiment; patterns reveal institutional order flow
Before completing your task, verify:
Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory.
import numpy as np
from typing import List, Dict, Tuple
from dataclasses import dataclass
from enum import Enum
class PatternType(Enum):
REVERSAL = "reversal"
CONTINUATION = "continuation"
consolidation = "consolidation"
@dataclass
class PatternResult:
pattern_type: PatternType
name: str
confirmation: float # 0-1 confidence score
trend_context: str
volume_profile: Dict[str, float]
class PricePatternDetector:
"""Detects candlestick and chart patterns across multiple timeframes."""
def __init__(self, lookback: int = 100):
self.lookback = lookback
self.min_strength = 0.6
def detect_all_patterns(
self, candles: np.ndarray, timeframe: str = "1h"
) -> List[PatternResult]:
"""Run all pattern detection algorithms on candle data."""
patterns = []
for pattern_func in [
self._detect_doji,
self._detect_hammer,
self._detect_engulfing,
self._detect_morning_star,
self._detect_head_and_shoulders,
]:
result = pattern_func(candles)
if result and result.confirmation >= self.min_strength:
patterns.append(result)
return patterns
def _detect_doji(self, candles: np.ndarray) -> PatternResult:
"""Detect Doji patterns indicating indecision."""
if len(candles) < 1:
return None
current = candles[-1]
body = abs(current['close'] - current['open'])
wick = current['high'] - current['low']
# Doji: body is very small relative to wick
body_ratio = body / wick if wick > 0 else 1
is_doji = body_ratio < 0.1
return PatternResult(
pattern_type=PatternType.REVERSAL,
name="doji" if is_doji else None,
confirmation=1 - body_ratio if is_doji else 0,
trend_context="neutral",
volume_profile=self._analyze_volume_profile(candles)
)
def _detect_hammer(self, candles: np.ndarray) -> PatternResult:
"""Detect Hammer pattern indicating bullish reversal."""
if len(candles) < 1:
return None
current = candles[-1]
body = abs(current['close'] - current['open'])
wick_upper = current['high'] - max(current['open'], current['close'])
wick_lower = min(current['open'], current['close']) - current['low']
# Hammer: small body, long lower wick, little upper wick
body_ratio = body / (wick_lower + 1e-8)
wick_ratio = wick_upper / (wick_lower + 1e-8)
is_hammer = body_ratio > 0.3 and wick_ratio < 0.5
return PatternResult(
pattern_type=PatternType.REVERSAL,
name="hammer" if is_hammer else None,
confirmation=body_ratio * (1 - wick_ratio) if is_hammer else 0,
trend_context="bullish" if is_hammer else "neutral",
volume_profile=self._analyze_volume_profile(candles)
)
def _detect_engulfing(self, candles: np.ndarray) -> PatternResult:
"""Detect Bullish/Bearish Engulfing patterns."""
if len(candles) < 2:
return None
prev, curr = candles[-2], candles[-1]
prev_body = prev['close'] - prev['open']
curr_body = curr['close'] - curr['open']
# Bullish Engulfing: previous bearish, current bullish and larger
is_bullish = prev_body < 0 and curr_body > 0 and abs(curr_body) > abs(prev_body)
# Bearish Engulfing: previous bullish, current bearish and larger
is_bearish = prev_body > 0 and curr_body < 0 and abs(curr_body) > abs(prev_body)
name = "bullish_engulfing" if is_bullish else "bearish_engulfing" if is_bearish else None
pattern_type = PatternType.REVERSAL if name else None
return PatternResult(
pattern_type=pattern_type,
name=name,
confirmation=abs(curr_body) / (abs(prev_body) + 1e-8) if name else 0,
trend_context="bullish" if is_bullish else "bearish" if is_bearish else "neutral",
volume_profile=self._analyze_volume_profile(candles)
)
def _detect_morning_star(self, candles: np.ndarray) -> PatternResult:
"""Detect Morning Star reversal pattern."""
if len(candles) < 3:
return None
c1, c2, c3 = candles[-3], candles[-2], candles[-1]
# Morning Star: bearish, small body (doji/spinning top), bullish
is_bearish_1 = c1['close'] < c1['open']
is_small_2 = abs(c2['close'] - c2['open']) < (c2['high'] - c2['low']) * 0.3
is_bullish_3 = c3['close'] > c3['open']
gap_up = c2['close'] < c3['open'] # Price gaps up
is_morning_star = is_bearish_1 and is_small_2 and is_bullish_3 and gap_up
return PatternResult(
pattern_type=PatternType.REVERSAL,
name="morning_star" if is_morning_star else None,
confirmation=0.8 if is_morning_star else 0,
trend_context="bullish" if is_morning_star else "neutral",
volume_profile=self._analyze_volume_profile(candles[-3:])
)
def _detect_head_and_shoulders(self, candles: np.ndarray) -> PatternResult:
"""Detect Head and Shoulders reversal pattern."""
if len(candles) < 5:
return None
# Simplified detection: find local maxima
highs = [(i, c['high']) for i, c in enumerate(candles[-5:])]
# Check for H&S structure: L-H-L-H-L
left Shoulder = highs[0][1]
head = max(highs[1][1], highs[3][1])
right_shoulder = highs[4][1]
is_hns = (
left_shoulder < head and
right_shoulder < head and
abs(left_shoulder - right_shoulder) / head < 0.1 # Symmetry
)
return PatternResult(
pattern_type=PatternType.REVERSAL,
name="head_and_shoulders" if is_hns else None,
confirmation=0.7 if is_hns else 0,
trend_context="bearish" if is_hns else "neutral",
volume_profile=self._analyze_volume_profile(candles[-5:])
)
def _analyze_volume_profile(self, candles: np.ndarray) -> Dict[str, float]:
"""Analyze volume characteristics."""
if len(candles) == 0:
return {"avg": 0, "std": 0, "current": 0}
volumes = [c['volume'] for c in candles]
return {
"avg": float(np.mean(volumes)),
"std": float(np.std(volumes)),
"current": float(volumes[-1]) if volumes else 0,
"volatility_ratio": float(np.std(volumes) / (np.mean(volumes) + 1e-8))
}
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Optional
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class TradeSignal:
"""Immutable trade signal with all required validation constraints."""
symbol: str
side: str # "buy" or "sell"
price: float
quantity: float
confidence: float # 0.0 to 1.0
reason: str
def validate(self) -> bool:
"""Validate that the trade signal meets all business constraints."""
if self.quantity <= 0:
raise ValueError(f"Quantity must be positive, got {self.quantity}")
if self.price <= 0:
raise ValueError(f"Price must be positive, got {self.price}")
if not 0.0 <= self.confidence <= 1.0:
raise ValueError(f"Confidence must be between 0 and 1, got {self.confidence}")
return True
def generate_trade_signal(
symbol: str,
side: str,
price: float,
quantity: float,
confidence: float,
reason: str,
) -> TradeSignal:
"""Generate a validated trade signal with guard clause checks."""
if side not in ("buy", "sell"):
raise ValueError(f"Invalid side '{side}', must be 'buy' or 'sell'")
signal = TradeSignal(
symbol=symbol,
side=side,
price=price,
quantity=quantity,
confidence=confidence,
reason=reason,
)
signal.validate()
logger.info("Trade signal generated: %s %s %.4f @ %.2f (confidence=%.2f)",
symbol, side, quantity, price, confidence)
return signal
def execute_with_risk_check(signal: TradeSignal, max_position_pct: float = 0.05) -> dict:
"""Execute a trade signal after applying risk management checks."""
adjusted_quantity = signal.quantity
if signal.side == "buy" and signal.quantity > max_position_pct:
logger.warning("Position %s exceeds max %.1f%% — capping to %.4f",
signal.symbol, max_position_pct * 100, max_position_pct)
adjusted_quantity = max_position_pct
return {
"symbol": signal.symbol,
"side": signal.side,
"price": signal.price,
"quantity": adjusted_quantity,
"capped": adjusted_quantity < signal.quantity,
"confidence": signal.confidence,
"status": "submitted",
}
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.