"""Pattern recognition model using learned and rule-based approaches.""" from __future__ import annotations import os from typing import List, Optional from core.candle_patterns import PatternSignal, PatternType, detect_all_patterns from core.trend_analysis import Candle try: import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import DataLoader, TensorDataset TORCH_AVAILABLE = True except ImportError: TORCH_AVAILABLE = False class PatternRecognizer: """ Hybrid pattern recognizer combining rule-based detection with a lightweight neural model for confirmation. Architecture (~15k parameters): - Input: normalized OHLCV + derived features of last 5 candles (8 features per candle) - Conv1d: 8 features -> 16 channels, kernel=3 - BatchNorm + ReLU - Conv1d: 16 -> 32 channels, kernel=3 - BatchNorm + ReLU - Conv1d: 32 -> 64 channels, kernel=3 - BatchNorm + ReLU - AdaptiveAvgPool1d -> 64 units - Linear -> 64 units, ReLU, Dropout(0.3) - Linear -> 32 units, ReLU, Dropout(0.3) - Linear -> num_patterns units (sigmoid for multi-label) """ PATTERN_COUNT = 10 # Number of recognized patterns PATTERN_NAMES = [ "bullish_engulfing", "bearish_engulfing", "morning_star", "evening_star", "hammer", "inverted_hammer", "shooting_star", "doji", "three_white_soldiers", "three_black_crows", ] def __init__(self, lookback: int = 5, model_path: Optional[str] = None): self.lookback = lookback self._model = None self._initialized = False self.model_path = model_path or os.path.join( os.path.dirname(os.path.dirname(__file__)), "models", "pattern_recognizer.pt" ) if TORCH_AVAILABLE: self._model = _PatternNet(lookback) self._initialized = True # Try to load saved weights if os.path.exists(self.model_path): try: self._model.load_state_dict(torch.load(self.model_path, weights_only=True, map_location="cpu")) self._model.eval() except Exception: pass def is_available(self) -> bool: """Check if PyTorch is available.""" return TORCH_AVAILABLE and self._initialized def detect(self, candles: List[Candle]) -> List[dict]: """ Detect patterns using both rule-based and ML approaches. Args: candles: List of candles (oldest to newest) Returns: List of detected patterns with type, confidence, and index """ results = [] # Rule-based detection (always available) rules = self._rule_based_detection(candles) results.extend(rules) # ML-based confirmation (if available) if self.is_available() and len(candles) >= self.lookback: ml_signals = self._ml_detection(candles) # Only add ML predictions that don't duplicate rule-based ml_types = {r["pattern"] for r in results} for signal in ml_signals: if signal["pattern"] not in ml_types and signal["confidence"] > 0.6: results.append(signal) return results def _rule_based_detection(self, candles: List[Candle]) -> List[dict]: """Use rule-based pattern detection from candle_patterns module.""" signals = detect_all_patterns(candles) results = [] for sig in signals: results.append({ "pattern": sig.pattern.value, "index": sig.index, "confidence": sig.confidence, "description": sig.description, "source": "rules", }) return results def _ml_detection(self, candles: List[Candle]) -> List[dict]: """ ML-based pattern detection using neural network. Returns list of detected patterns with confidence > 0.5. """ if not self.is_available(): return [] features = self._extract_features(candles) self._model.eval() with torch.no_grad(): features_tensor = torch.FloatTensor(features).reshape(8, self.lookback).unsqueeze(0) output = self._model(features_tensor) probabilities = torch.sigmoid(output) results = [] for i, prob in enumerate(probabilities[0]): if prob.item() > 0.5: results.append({ "pattern": self.PATTERN_NAMES[i], "index": len(candles) - 1, "confidence": round(prob.item(), 4), "description": f"ML detected {self.PATTERN_NAMES[i]}", "source": "ml", }) return results def _extract_features(self, candles: List[Candle]) -> List[float]: """ Extract normalized features from the last `lookback` candles. Returns flat list of length lookback * 8. Features per candle: open_ratio, high_ratio, low_ratio, close_ratio, body_ratio, upper_shadow_ratio, lower_shadow_ratio, volume_ratio """ recent = candles[-self.lookback:] if not recent: return [] closes = [c.close for c in recent] price_min = min(closes) price_max = max(closes) price_range = price_max - price_min if price_max != price_min else 1.0 volumes = [c.volume for c in recent] vol_min = min(volumes) vol_max = max(volumes) vol_range = vol_max - vol_min if vol_max != vol_min else 1.0 features = [] for c in recent: body = abs(c.close - c.open) total_range = c.high - c.low if c.high != c.low else 1.0 upper = c.high - max(c.open, c.close) lower = min(c.open, c.close) - c.low features.append((c.open - price_min) / price_range) features.append((c.high - price_min) / price_range) features.append((c.low - price_min) / price_range) features.append((c.close - price_min) / price_range) features.append(body / total_range) features.append(upper / total_range) features.append(lower / total_range) features.append((c.volume - vol_min) / vol_range) return features def train(self, training_data: List[List[Candle]], pattern_labels: List[List[int]], epochs: int = 50, lr: float = 0.001, save: bool = True) -> dict: """ Train the pattern recognizer. Args: training_data: List of candle sequences pattern_labels: List of binary label lists per pattern per sequence epochs: Number of training epochs lr: Learning rate save: Whether to save model weights after training Returns: Training history """ if not self.is_available(): return {"error": "PyTorch not available"} if len(training_data) != len(pattern_labels): return {"error": "Data and labels length mismatch"} # Filter sequences that are long enough valid_features = [] valid_labels = [] for i, candles in enumerate(training_data): if len(candles) >= self.lookback: valid_features.append(self._extract_features(candles)) valid_labels.append(pattern_labels[i]) if not valid_features: return {"error": "No valid training sequences"} features_tensor = torch.FloatTensor(valid_features) labels_tensor = torch.FloatTensor(valid_labels) batch_size = min(64, len(valid_features)) # Compute class weights for imbalance (patterns with fewer positives get higher weight) pos_counts = labels_tensor.sum(dim=0) n_samples = len(valid_features) # Weight inversely proportional to positive count, with a floor class_weights = n_samples / (self.PATTERN_COUNT * pos_counts.clamp(min=1)) class_weights = class_weights / class_weights.mean() # Normalize self._model.train() optimizer = torch.optim.Adam(self._model.parameters(), lr=lr, weight_decay=1e-4) criterion = nn.BCEWithLogitsLoss(pos_weight=class_weights) dataset = TensorDataset(features_tensor, labels_tensor) loader = DataLoader(dataset, batch_size=batch_size, shuffle=True) history = {"loss": [], "accuracy": []} for epoch in range(epochs): epoch_loss = 0.0 epoch_correct = 0 epoch_total = 0 optimizer.zero_grad() batches = 0 for batch_features, batch_labels in loader: # Reshape features from (N, lookback*5) to (N, 5, lookback) for Conv1d batch_input = batch_features.reshape(len(batch_features), 8, self.lookback) outputs = self._model(batch_input) loss = criterion(outputs, batch_labels) loss.backward() optimizer.step() optimizer.zero_grad() epoch_loss += loss.item() batches += 1 with torch.no_grad(): predicted = (torch.sigmoid(outputs) > 0.5).float() correct = (predicted == batch_labels).float().sum(dim=1) epoch_correct += (correct == self.PATTERN_COUNT).sum().item() epoch_total += len(batch_features) avg_loss = epoch_loss / (batches or 1) accuracy = epoch_correct / (epoch_total or 1) history["loss"].append(round(avg_loss, 6)) history["accuracy"].append(round(accuracy, 4)) result = { "final_loss": round(history["loss"][-1], 6), "final_accuracy": round(history["accuracy"][-1], 4), "epochs_trained": epochs, "total_samples": len(valid_features), } if save: self.save_model() return result def save_model(self) -> bool: """Save model weights to disk.""" if not self.is_available(): return False try: os.makedirs(os.path.dirname(self.model_path), exist_ok=True) torch.save(self._model.state_dict(), self.model_path) return True except Exception: return False def load_model(self) -> bool: """Load model weights from disk.""" if not self.is_available() or not os.path.exists(self.model_path): return False try: self._model.load_state_dict(torch.load(self.model_path, weights_only=True, map_location="cpu")) self._model.eval() return True except Exception: return False class _PatternNet(nn.Module): """Internal PyTorch model for multi-label pattern classification.""" def __init__(self, lookback: int, num_patterns: int = 10): super().__init__() # Input: (batch, 8 features, lookback) self.conv1 = nn.Conv1d(8, 16, kernel_size=3, padding=1) self.bn1 = nn.BatchNorm1d(16) self.conv2 = nn.Conv1d(16, 32, kernel_size=3, padding=1) self.bn2 = nn.BatchNorm1d(32) self.conv3 = nn.Conv1d(32, 64, kernel_size=3, padding=1) self.bn3 = nn.BatchNorm1d(64) self.pool = nn.AdaptiveAvgPool1d(1) self.fc1 = nn.Linear(64, 64) self.fc2 = nn.Linear(64, 32) self.fc3 = nn.Linear(32, num_patterns) self.relu = nn.ReLU() self.dropout = nn.Dropout(0.3) def forward(self, x): # x shape: (batch, 8, lookback) x = self.relu(self.bn1(self.conv1(x))) x = self.relu(self.bn2(self.conv2(x))) x = self.relu(self.bn3(self.conv3(x))) x = self.pool(x).squeeze(-1) x = self.dropout(self.relu(self.fc1(x))) x = self.dropout(self.relu(self.fc2(x))) x = self.fc3(x) return x