Spaces:
Sleeping
Sleeping
| """ | |
| Candlestick chart generation using mplfinance. | |
| This module provides functionality to generate candlestick charts with volume, | |
| save them to files, and optionally overlay technical indicators. | |
| """ | |
| import os | |
| from datetime import datetime | |
| from pathlib import Path | |
| from typing import Any, Dict, List, Optional, Tuple | |
| import matplotlib.pyplot as plt | |
| import mplfinance as mpf | |
| import pandas as pd | |
| from matplotlib.figure import Figure | |
| class ChartGenerationError(Exception): | |
| """Raised when chart generation fails.""" | |
| pass | |
| class ChartGenerator: | |
| """Generates candlestick charts using mplfinance.""" | |
| DEFAULT_STYLE = "charles" | |
| DEFAULT_CHART_TYPE = "candle" | |
| DEFAULT_VOLUME = True | |
| DEFAULT_FIGSIZE = (12, 8) | |
| DEFAULT_DPI = 100 | |
| def __init__( | |
| self, | |
| output_dir: str = "data/charts", | |
| style: str = DEFAULT_STYLE, | |
| figsize: Tuple[int, int] = DEFAULT_FIGSIZE, | |
| dpi: int = DEFAULT_DPI, | |
| ): | |
| """ | |
| Initialize chart generator. | |
| Args: | |
| output_dir: Directory to save generated charts | |
| style: mplfinance style (charles, yahoo, binance, etc.) | |
| figsize: Figure size as (width, height) tuple | |
| dpi: Dots per inch for saved images | |
| """ | |
| self.output_dir = Path(output_dir) | |
| self.style = style | |
| self.figsize = figsize | |
| self.dpi = dpi | |
| self.output_dir.mkdir(parents=True, exist_ok=True) | |
| def generate_candlestick_chart( | |
| self, | |
| df: pd.DataFrame, | |
| ticker: str, | |
| timeframe: str, | |
| title: Optional[str] = None, | |
| volume: bool = DEFAULT_VOLUME, | |
| show_nontrading: bool = False, | |
| save: bool = True, | |
| filename: Optional[str] = None, | |
| indicators: Optional[List[Dict[str, Any]]] = None, | |
| ) -> Tuple[Figure, Optional[str]]: | |
| """ | |
| Generate candlestick chart from OHLC data. | |
| Args: | |
| df: DataFrame with DatetimeIndex and OHLC columns (Open, High, Low, Close, Volume) | |
| ticker: Stock ticker symbol | |
| timeframe: Timeframe (1m, 5m, 1h, 1d, etc.) | |
| title: Custom chart title (default: auto-generated) | |
| volume: Whether to show volume panel | |
| show_nontrading: Whether to show non-trading periods (gaps) | |
| save: Whether to save chart to file | |
| filename: Custom filename (default: auto-generated) | |
| indicators: List of indicator overlays (see add_indicators for format) | |
| Returns: | |
| Tuple of (matplotlib Figure, filepath if saved) | |
| Raises: | |
| ChartGenerationError: If chart generation fails | |
| """ | |
| try: | |
| # Check if chart already exists (caching for performance) | |
| if save and filename: | |
| cached_path = self.output_dir / filename | |
| if not cached_path.name.endswith(".png"): | |
| cached_path = self.output_dir / (filename + ".png") | |
| if cached_path.exists(): | |
| # Return None for figure since we're using cached version | |
| return None, str(cached_path) | |
| # Validate DataFrame | |
| self._validate_ohlc_dataframe(df) | |
| # Prepare DataFrame for mplfinance | |
| df_plot = df.copy() | |
| # mplfinance requires DatetimeIndex - set timestamp as index if it's a column | |
| if "timestamp" in df_plot.columns: | |
| df_plot["timestamp"] = pd.to_datetime(df_plot["timestamp"]) | |
| df_plot = df_plot.set_index("timestamp") | |
| # mplfinance requires capitalized column names | |
| column_mapping = { | |
| "open": "Open", | |
| "high": "High", | |
| "low": "Low", | |
| "close": "Close", | |
| "volume": "Volume", | |
| } | |
| df_plot = df_plot.rename(columns=column_mapping) | |
| # Generate title | |
| if title is None: | |
| title = f"{ticker.upper()} - {timeframe} Candlestick Chart" | |
| # Prepare plot arguments | |
| kwargs = { | |
| "type": self.DEFAULT_CHART_TYPE, | |
| "style": self.style, | |
| "title": title, | |
| "volume": volume, | |
| "figsize": self.figsize, | |
| "show_nontrading": show_nontrading, | |
| "returnfig": True, | |
| } | |
| # Add indicators if provided | |
| if indicators: | |
| addplot = self._prepare_indicator_overlays(df, indicators) | |
| if addplot: | |
| kwargs["addplot"] = addplot | |
| # Generate chart | |
| fig, axes = mpf.plot(df_plot, **kwargs) | |
| # Save to file if requested | |
| filepath = None | |
| if save: | |
| filepath = self._save_chart(fig, ticker, timeframe, filename) | |
| return fig, filepath | |
| except Exception as e: | |
| raise ChartGenerationError( | |
| f"Failed to generate chart for {ticker}: {str(e)}" | |
| ) from e | |
| def generate_comparison_chart( | |
| self, | |
| df: pd.DataFrame, | |
| ticker: str, | |
| timeframe: str, | |
| indicator_data: Dict[str, pd.Series], | |
| title: Optional[str] = None, | |
| save: bool = True, | |
| filename: Optional[str] = None, | |
| ) -> Tuple[Figure, Optional[str]]: | |
| """ | |
| Generate chart with multiple indicator overlays for comparison. | |
| Args: | |
| df: OHLC DataFrame | |
| ticker: Stock ticker | |
| timeframe: Timeframe | |
| indicator_data: Dict mapping indicator names to Series data | |
| title: Custom title | |
| save: Whether to save | |
| filename: Custom filename | |
| Returns: | |
| Tuple of (Figure, filepath) | |
| """ | |
| indicators = [] | |
| for name, series in indicator_data.items(): | |
| indicators.append( | |
| { | |
| "data": series, | |
| "panel": 0, # Overlay on main chart | |
| "ylabel": name, | |
| "secondary_y": False, | |
| } | |
| ) | |
| return self.generate_candlestick_chart( | |
| df=df, | |
| ticker=ticker, | |
| timeframe=timeframe, | |
| title=title, | |
| save=save, | |
| filename=filename, | |
| indicators=indicators, | |
| ) | |
| def _validate_ohlc_dataframe(self, df: pd.DataFrame) -> None: | |
| """ | |
| Validate that DataFrame has required OHLC columns and DatetimeIndex. | |
| Raises: | |
| ChartGenerationError: If validation fails | |
| """ | |
| # Check if DataFrame is empty FIRST before checking columns | |
| if df.empty: | |
| raise ChartGenerationError("DataFrame is empty - no market data available") | |
| required_columns = ["open", "high", "low", "close"] | |
| missing_columns = [col for col in required_columns if col not in df.columns] | |
| if missing_columns: | |
| raise ChartGenerationError( | |
| f"DataFrame missing required columns: {missing_columns}. Available columns: {list(df.columns)}" | |
| ) | |
| # Check for timestamp - either as index or as column | |
| if not isinstance(df.index, pd.DatetimeIndex) and "timestamp" not in df.columns: | |
| raise ChartGenerationError( | |
| "DataFrame must have DatetimeIndex or timestamp column" | |
| ) | |
| def _prepare_indicator_overlays( | |
| self, | |
| df: pd.DataFrame, | |
| indicators: List[Dict[str, Any]], | |
| ) -> List[mpf.make_addplot]: | |
| """ | |
| Prepare indicator data for mplfinance addplot. | |
| Args: | |
| df: OHLC DataFrame | |
| indicators: List of dicts with keys: | |
| - data: pd.Series with indicator values | |
| - panel: 0 for main chart, 1+ for separate panels | |
| - color: Line color (optional) | |
| - width: Line width (optional) | |
| - ylabel: Y-axis label (optional) | |
| - secondary_y: Use secondary y-axis (optional) | |
| Returns: | |
| List of mplfinance addplot objects | |
| """ | |
| addplot = [] | |
| for ind in indicators: | |
| plot_kwargs = { | |
| "panel": ind.get("panel", 0), | |
| "ylabel": ind.get("ylabel", ""), | |
| "secondary_y": ind.get("secondary_y", False), | |
| } | |
| if "color" in ind: | |
| plot_kwargs["color"] = ind["color"] | |
| if "width" in ind: | |
| plot_kwargs["width"] = ind["width"] | |
| addplot.append(mpf.make_addplot(ind["data"], **plot_kwargs)) | |
| return addplot | |
| def _save_chart( | |
| self, | |
| fig: Figure, | |
| ticker: str, | |
| timeframe: str, | |
| filename: Optional[str] = None, | |
| ) -> str: | |
| """ | |
| Save chart to file. | |
| Args: | |
| fig: Matplotlib figure | |
| ticker: Stock ticker | |
| timeframe: Timeframe | |
| filename: Custom filename (default: auto-generated) | |
| Returns: | |
| Filepath where chart was saved | |
| """ | |
| if filename is None: | |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") | |
| filename = f"{ticker}_{timeframe}_{timestamp}.png" | |
| # Ensure .png extension | |
| if not filename.endswith(".png"): | |
| filename += ".png" | |
| filepath = self.output_dir / filename | |
| # Save figure (matplotlib handles PNG compression internally) | |
| fig.savefig( | |
| filepath, | |
| dpi=self.dpi, | |
| bbox_inches="tight", | |
| ) | |
| return str(filepath) | |
| def generate_rsi_chart( | |
| self, | |
| df: pd.DataFrame, | |
| rsi_series: pd.Series, | |
| ticker: str, | |
| timeframe: str, | |
| rsi_period: int = 14, | |
| save: bool = True, | |
| filename: Optional[str] = None, | |
| ) -> Tuple[Figure, Optional[str]]: | |
| """ | |
| Generate RSI chart with overbought/oversold zones. | |
| Args: | |
| df: OHLC DataFrame | |
| rsi_series: RSI values as Series | |
| ticker: Stock ticker | |
| timeframe: Timeframe | |
| rsi_period: RSI period (for title) | |
| save: Whether to save chart | |
| filename: Custom filename | |
| Returns: | |
| Tuple of (Figure, filepath) | |
| """ | |
| try: | |
| # Validate data | |
| self._validate_ohlc_dataframe(df) | |
| # Create figure with single subplot for RSI only | |
| fig, ax = plt.subplots( | |
| 1, 1, figsize=(self.figsize[0], self.figsize[1] * 0.4) | |
| ) | |
| # Plot RSI | |
| ax.plot( | |
| rsi_series.index, | |
| rsi_series.values, | |
| color="#2962FF", | |
| linewidth=2, | |
| label=f"RSI({rsi_period})", | |
| ) | |
| # Add overbought/oversold zones | |
| ax.axhline( | |
| y=70, | |
| color="#F44336", | |
| linestyle="--", | |
| linewidth=1, | |
| alpha=0.7, | |
| label="Overbought (70)", | |
| ) | |
| ax.axhline( | |
| y=30, | |
| color="#4CAF50", | |
| linestyle="--", | |
| linewidth=1, | |
| alpha=0.7, | |
| label="Oversold (30)", | |
| ) | |
| ax.fill_between(rsi_series.index, 70, 100, alpha=0.1, color="#F44336") | |
| ax.fill_between(rsi_series.index, 0, 30, alpha=0.1, color="#4CAF50") | |
| # Add neutral zone | |
| ax.axhline(y=50, color="gray", linestyle=":", linewidth=0.5, alpha=0.5) | |
| # Format RSI subplot | |
| ax.set_ylim(0, 100) | |
| ax.set_ylabel("RSI", fontsize=10) | |
| ax.set_xlabel("Date", fontsize=10) | |
| ax.legend(loc="upper left", fontsize=8) | |
| ax.grid(True, alpha=0.3) | |
| # Set title | |
| fig.suptitle( | |
| f"{ticker.upper()} - RSI({rsi_period}) Analysis ({timeframe})", | |
| fontsize=12, | |
| fontweight="bold", | |
| ) | |
| plt.tight_layout() | |
| # Save if requested | |
| filepath = None | |
| if save: | |
| if filename is None: | |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") | |
| filename = f"{ticker}_rsi_{timeframe}_{timestamp}.png" | |
| filepath = self._save_chart(fig, ticker, timeframe, filename) | |
| return fig, filepath | |
| except Exception as e: | |
| raise ChartGenerationError(f"Failed to generate RSI chart: {str(e)}") from e | |
| def generate_macd_chart( | |
| self, | |
| df: pd.DataFrame, | |
| macd: pd.Series, | |
| signal: pd.Series, | |
| histogram: pd.Series, | |
| ticker: str, | |
| timeframe: str, | |
| save: bool = True, | |
| filename: Optional[str] = None, | |
| ) -> Tuple[Figure, Optional[str]]: | |
| """ | |
| Generate MACD chart with signal line and histogram. | |
| Args: | |
| df: OHLC DataFrame | |
| macd: MACD line values | |
| signal: Signal line values | |
| histogram: MACD histogram values | |
| ticker: Stock ticker | |
| timeframe: Timeframe | |
| save: Whether to save chart | |
| filename: Custom filename | |
| Returns: | |
| Tuple of (Figure, filepath) | |
| """ | |
| try: | |
| # Validate data | |
| self._validate_ohlc_dataframe(df) | |
| # Create figure with single subplot for MACD only | |
| fig, ax = plt.subplots( | |
| 1, 1, figsize=(self.figsize[0], self.figsize[1] * 0.4) | |
| ) | |
| # Plot MACD | |
| ax.plot(macd.index, macd.values, color="#2962FF", linewidth=2, label="MACD") | |
| ax.plot( | |
| signal.index, | |
| signal.values, | |
| color="#FF6D00", | |
| linewidth=2, | |
| label="Signal", | |
| ) | |
| # Plot histogram with colors | |
| colors = ["#4CAF50" if h >= 0 else "#F44336" for h in histogram.values] | |
| ax.bar( | |
| histogram.index, | |
| histogram.values, | |
| color=colors, | |
| alpha=0.3, | |
| label="Histogram", | |
| ) | |
| # Add zero line | |
| ax.axhline(y=0, color="gray", linestyle="-", linewidth=0.5, alpha=0.5) | |
| # Format MACD subplot | |
| ax.set_ylabel("MACD", fontsize=10) | |
| ax.set_xlabel("Date", fontsize=10) | |
| ax.legend(loc="upper left", fontsize=8) | |
| ax.grid(True, alpha=0.3) | |
| # Set title | |
| fig.suptitle( | |
| f"{ticker.upper()} - MACD Analysis ({timeframe})", | |
| fontsize=12, | |
| fontweight="bold", | |
| ) | |
| plt.tight_layout() | |
| # Save if requested | |
| filepath = None | |
| if save: | |
| if filename is None: | |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") | |
| filename = f"{ticker}_macd_{timeframe}_{timestamp}.png" | |
| filepath = self._save_chart(fig, ticker, timeframe, filename) | |
| return fig, filepath | |
| except Exception as e: | |
| raise ChartGenerationError( | |
| f"Failed to generate MACD chart: {str(e)}" | |
| ) from e | |
| def generate_stochastic_chart( | |
| self, | |
| df: pd.DataFrame, | |
| k_series: pd.Series, | |
| d_series: pd.Series, | |
| ticker: str, | |
| timeframe: str, | |
| save: bool = True, | |
| filename: Optional[str] = None, | |
| ) -> Tuple[Figure, Optional[str]]: | |
| """ | |
| Generate Stochastic Oscillator chart. | |
| Args: | |
| df: OHLC DataFrame | |
| k_series: %K line values | |
| d_series: %D line values | |
| ticker: Stock ticker | |
| timeframe: Timeframe | |
| save: Whether to save chart | |
| filename: Custom filename | |
| Returns: | |
| Tuple of (Figure, filepath) | |
| """ | |
| try: | |
| # Validate data | |
| self._validate_ohlc_dataframe(df) | |
| # Create figure with single subplot for Stochastic only | |
| fig, ax = plt.subplots( | |
| 1, 1, figsize=(self.figsize[0], self.figsize[1] * 0.4) | |
| ) | |
| # Plot Stochastic | |
| ax.plot( | |
| k_series.index, | |
| k_series.values, | |
| color="#2962FF", | |
| linewidth=2, | |
| label="%K", | |
| ) | |
| ax.plot( | |
| d_series.index, | |
| d_series.values, | |
| color="#FF6D00", | |
| linewidth=2, | |
| label="%D", | |
| ) | |
| # Add overbought/oversold zones | |
| ax.axhline( | |
| y=80, | |
| color="#F44336", | |
| linestyle="--", | |
| linewidth=1, | |
| alpha=0.7, | |
| label="Overbought (80)", | |
| ) | |
| ax.axhline( | |
| y=20, | |
| color="#4CAF50", | |
| linestyle="--", | |
| linewidth=1, | |
| alpha=0.7, | |
| label="Oversold (20)", | |
| ) | |
| ax.fill_between(k_series.index, 80, 100, alpha=0.1, color="#F44336") | |
| ax.fill_between(k_series.index, 0, 20, alpha=0.1, color="#4CAF50") | |
| # Format Stochastic subplot | |
| ax.set_ylim(0, 100) | |
| ax.set_ylabel("Stochastic", fontsize=10) | |
| ax.set_xlabel("Date", fontsize=10) | |
| ax.legend(loc="upper left", fontsize=8) | |
| ax.grid(True, alpha=0.3) | |
| # Set title | |
| fig.suptitle( | |
| f"{ticker.upper()} - Stochastic Oscillator ({timeframe})", | |
| fontsize=12, | |
| fontweight="bold", | |
| ) | |
| plt.tight_layout() | |
| # Save if requested | |
| filepath = None | |
| if save: | |
| if filename is None: | |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") | |
| filename = f"{ticker}_stochastic_{timeframe}_{timestamp}.png" | |
| filepath = self._save_chart(fig, ticker, timeframe, filename) | |
| return fig, filepath | |
| except Exception as e: | |
| raise ChartGenerationError( | |
| f"Failed to generate Stochastic chart: {str(e)}" | |
| ) from e | |
| def generate_candlestick_pattern_history_chart( | |
| self, | |
| df: pd.DataFrame, | |
| ticker: str, | |
| timeframe: str, | |
| patterns: List[Dict[str, Any]], | |
| save: bool = True, | |
| filename: Optional[str] = None, | |
| ) -> Tuple[Figure, Optional[str]]: | |
| """ | |
| Generate candlestick chart with historical pattern markers. | |
| Args: | |
| df: DataFrame with DatetimeIndex and OHLC columns | |
| ticker: Stock ticker symbol | |
| timeframe: Timeframe string | |
| patterns: List of detected candlestick patterns with date/index info | |
| save: Whether to save chart to file | |
| filename: Custom filename | |
| Returns: | |
| Tuple of (Figure, filepath) | |
| """ | |
| try: | |
| # Ensure DatetimeIndex for mplfinance | |
| if not isinstance(df.index, pd.DatetimeIndex): | |
| df = df.copy() | |
| df.index = pd.to_datetime(df.index) | |
| # Create figure with candlestick chart | |
| fig, axes = mpf.plot( | |
| df, | |
| type=self.DEFAULT_CHART_TYPE, | |
| style=self.style, | |
| volume=True, | |
| figsize=self.figsize, | |
| returnfig=True, | |
| ylabel="Price", | |
| ylabel_lower="Volume", | |
| ) | |
| # Add pattern markers | |
| ax = axes[0] # Main price chart axis | |
| # Group patterns by date and add markers | |
| for pattern in patterns[:20]: # Limit to 20 most recent patterns | |
| pattern_date = pattern.get("date") | |
| pattern_name = pattern.get("pattern", "Unknown") | |
| if pattern_date: | |
| try: | |
| # Convert to datetime if string | |
| if isinstance(pattern_date, str): | |
| pattern_date = pd.to_datetime(pattern_date) | |
| # Find closest date in dataframe | |
| if pattern_date in df.index: | |
| idx = df.index.get_loc(pattern_date) | |
| price = df.iloc[idx]["high"] | |
| # Add annotation with arrow | |
| ax.annotate( | |
| pattern_name, | |
| xy=(idx, price), | |
| xytext=(idx, price * 1.02), | |
| fontsize=7, | |
| ha="center", | |
| bbox=dict( | |
| boxstyle="round,pad=0.3", | |
| facecolor="yellow", | |
| alpha=0.7, | |
| ), | |
| arrowprops=dict(arrowstyle="->", color="black", lw=0.5), | |
| ) | |
| except Exception as e: | |
| continue | |
| # Set title | |
| pattern_count = len(patterns) | |
| fig.suptitle( | |
| f"{ticker.upper()} - {pattern_count} Candlestick Patterns ({timeframe})", | |
| fontsize=12, | |
| fontweight="bold", | |
| ) | |
| # Save if requested | |
| filepath = None | |
| if save: | |
| if filename is None: | |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") | |
| filename = f"{ticker}_pattern_history_{timeframe}_{timestamp}.png" | |
| filepath = self._save_chart(fig, ticker, timeframe, filename) | |
| return fig, filepath | |
| except Exception as e: | |
| raise ChartGenerationError( | |
| f"Failed to generate pattern history chart: {str(e)}" | |
| ) from e | |
| def generate_pattern_annotated_chart( | |
| self, | |
| df: pd.DataFrame, | |
| ticker: str, | |
| timeframe: str, | |
| pattern: Dict[str, Any], | |
| save: bool = True, | |
| filename: Optional[str] = None, | |
| ) -> Tuple[Figure, Optional[str]]: | |
| """ | |
| Generate candlestick chart with pattern annotations. | |
| Args: | |
| df: DataFrame with DatetimeIndex and OHLC columns | |
| ticker: Stock ticker symbol | |
| timeframe: Timeframe string | |
| pattern: Pattern dict with type, points, and metadata | |
| save: Whether to save chart to file | |
| filename: Custom filename | |
| Returns: | |
| Tuple of (Figure, filepath) | |
| """ | |
| try: | |
| # Ensure DatetimeIndex for mplfinance | |
| if not isinstance(df.index, pd.DatetimeIndex): | |
| df = df.copy() | |
| df.index = pd.to_datetime(df.index) | |
| # Create figure with candlestick chart | |
| fig, axes = mpf.plot( | |
| df, | |
| type=self.DEFAULT_CHART_TYPE, | |
| style=self.style, | |
| volume=True, | |
| figsize=self.figsize, | |
| returnfig=True, | |
| ylabel="Price", | |
| ylabel_lower="Volume", | |
| ) | |
| ax = axes[0] # Main price axis | |
| # Get pattern type and add appropriate annotations | |
| pattern_type = pattern.get("type", "").lower() | |
| if "head" in pattern_type and "shoulder" in pattern_type: | |
| self._annotate_head_and_shoulders(ax, df, pattern) | |
| elif "double" in pattern_type and ( | |
| "bottom" in pattern_type or "top" in pattern_type | |
| ): | |
| self._annotate_double_bottom_top(ax, df, pattern) | |
| elif "triangle" in pattern_type: | |
| self._annotate_triangle(ax, df, pattern) | |
| # Set title with pattern name | |
| pattern_name = pattern.get("type", "Pattern") | |
| signal = pattern.get("signal", "").upper() | |
| fig.suptitle( | |
| f"{ticker.upper()} - {pattern_name} Pattern ({signal}) - {timeframe}", | |
| fontsize=12, | |
| fontweight="bold", | |
| ) | |
| plt.tight_layout() | |
| # Save if requested | |
| filepath = None | |
| if save: | |
| if filename is None: | |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") | |
| pattern_slug = pattern_name.lower().replace(" ", "_") | |
| filename = f"{ticker}_{pattern_slug}_{timeframe}_{timestamp}.png" | |
| filepath = self._save_chart(fig, ticker, timeframe, filename) | |
| return fig, filepath | |
| except Exception as e: | |
| raise ChartGenerationError( | |
| f"Failed to generate pattern annotated chart: {str(e)}" | |
| ) from e | |
| def _annotate_head_and_shoulders( | |
| self, ax: plt.Axes, df: pd.DataFrame, pattern: Dict[str, Any] | |
| ) -> None: | |
| """ | |
| Annotate head-and-shoulders pattern on chart. | |
| Args: | |
| ax: Matplotlib axes | |
| df: OHLC DataFrame | |
| pattern: Pattern dict with shoulder/head indices | |
| """ | |
| points = pattern.get("points", {}) | |
| left_shoulder_idx = points.get("left_shoulder") | |
| head_idx = points.get("head") | |
| right_shoulder_idx = points.get("right_shoulder") | |
| neckline = points.get("neckline", []) | |
| if not all( | |
| [ | |
| left_shoulder_idx is not None, | |
| head_idx is not None, | |
| right_shoulder_idx is not None, | |
| ] | |
| ): | |
| return | |
| # Get price values | |
| left_shoulder_price = df.iloc[left_shoulder_idx]["high"] | |
| head_price = df.iloc[head_idx]["high"] | |
| right_shoulder_price = df.iloc[right_shoulder_idx]["high"] | |
| # Annotate shoulders and head | |
| ax.annotate( | |
| "Left\nShoulder", | |
| xy=(left_shoulder_idx, left_shoulder_price), | |
| xytext=(left_shoulder_idx, left_shoulder_price * 1.05), | |
| fontsize=9, | |
| ha="center", | |
| bbox=dict(boxstyle="round,pad=0.3", facecolor="yellow", alpha=0.7), | |
| arrowprops=dict(arrowstyle="->", color="black", lw=1.5), | |
| ) | |
| ax.annotate( | |
| "Head", | |
| xy=(head_idx, head_price), | |
| xytext=(head_idx, head_price * 1.05), | |
| fontsize=10, | |
| fontweight="bold", | |
| ha="center", | |
| bbox=dict(boxstyle="round,pad=0.3", facecolor="red", alpha=0.7), | |
| arrowprops=dict(arrowstyle="->", color="black", lw=2), | |
| ) | |
| ax.annotate( | |
| "Right\nShoulder", | |
| xy=(right_shoulder_idx, right_shoulder_price), | |
| xytext=(right_shoulder_idx, right_shoulder_price * 1.05), | |
| fontsize=9, | |
| ha="center", | |
| bbox=dict(boxstyle="round,pad=0.3", facecolor="yellow", alpha=0.7), | |
| arrowprops=dict(arrowstyle="->", color="black", lw=1.5), | |
| ) | |
| # Draw neckline if provided | |
| if neckline and len(neckline) >= 2: | |
| neckline_x = [point[0] for point in neckline] | |
| neckline_y = [point[1] for point in neckline] | |
| ax.plot( | |
| neckline_x, | |
| neckline_y, | |
| "b--", | |
| linewidth=2, | |
| label="Neckline", | |
| alpha=0.8, | |
| ) | |
| ax.legend(loc="best", fontsize=8) | |
| def _annotate_double_bottom_top( | |
| self, ax: plt.Axes, df: pd.DataFrame, pattern: Dict[str, Any] | |
| ) -> None: | |
| """ | |
| Annotate double-bottom or double-top pattern on chart. | |
| Args: | |
| ax: Matplotlib axes | |
| df: OHLC DataFrame | |
| pattern: Pattern dict with first/second bottom/top indices | |
| """ | |
| points = pattern.get("points", {}) | |
| is_double_bottom = "bottom" in pattern.get("type", "").lower() | |
| first_idx = points.get("first") | |
| second_idx = points.get("second") | |
| resistance_support = points.get("resistance_support") | |
| if first_idx is None or second_idx is None: | |
| return | |
| # Get prices | |
| if is_double_bottom: | |
| first_price = df.iloc[first_idx]["low"] | |
| second_price = df.iloc[second_idx]["low"] | |
| label_text = "Bottom" | |
| color = "green" | |
| else: | |
| first_price = df.iloc[first_idx]["high"] | |
| second_price = df.iloc[second_idx]["high"] | |
| label_text = "Top" | |
| color = "red" | |
| # Annotate first and second bottom/top | |
| ax.annotate( | |
| f"First\n{label_text}", | |
| xy=(first_idx, first_price), | |
| xytext=(first_idx, first_price * (0.95 if is_double_bottom else 1.05)), | |
| fontsize=9, | |
| ha="center", | |
| bbox=dict(boxstyle="round,pad=0.3", facecolor=color, alpha=0.7), | |
| arrowprops=dict(arrowstyle="->", color="black", lw=1.5), | |
| ) | |
| ax.annotate( | |
| f"Second\n{label_text}", | |
| xy=(second_idx, second_price), | |
| xytext=(second_idx, second_price * (0.95 if is_double_bottom else 1.05)), | |
| fontsize=9, | |
| ha="center", | |
| bbox=dict(boxstyle="round,pad=0.3", facecolor=color, alpha=0.7), | |
| arrowprops=dict(arrowstyle="->", color="black", lw=1.5), | |
| ) | |
| # Draw resistance/support line if provided | |
| if resistance_support is not None: | |
| ax.axhline( | |
| y=resistance_support, | |
| color="blue", | |
| linestyle="--", | |
| linewidth=2, | |
| label="Resistance" if is_double_bottom else "Support", | |
| alpha=0.8, | |
| ) | |
| ax.legend(loc="best", fontsize=8) | |
| def _annotate_triangle( | |
| self, ax: plt.Axes, df: pd.DataFrame, pattern: Dict[str, Any] | |
| ) -> None: | |
| """ | |
| Annotate triangle pattern on chart. | |
| Args: | |
| ax: Matplotlib axes | |
| df: OHLC DataFrame | |
| pattern: Pattern dict with trendline data | |
| """ | |
| points = pattern.get("points", {}) | |
| upper_trendline = points.get("upper_trendline", []) | |
| lower_trendline = points.get("lower_trendline", []) | |
| pattern_type = pattern.get("type", "").lower() | |
| # Draw upper trendline | |
| if upper_trendline and len(upper_trendline) >= 2: | |
| upper_x = [point[0] for point in upper_trendline] | |
| upper_y = [point[1] for point in upper_trendline] | |
| if "ascending" in pattern_type: | |
| ax.plot( | |
| upper_x, upper_y, "r--", linewidth=2, label="Resistance", alpha=0.8 | |
| ) | |
| else: | |
| ax.plot( | |
| upper_x, | |
| upper_y, | |
| "r-", | |
| linewidth=2, | |
| label="Falling Resistance", | |
| alpha=0.8, | |
| ) | |
| # Draw lower trendline | |
| if lower_trendline and len(lower_trendline) >= 2: | |
| lower_x = [point[0] for point in lower_trendline] | |
| lower_y = [point[1] for point in lower_trendline] | |
| if "ascending" in pattern_type: | |
| ax.plot( | |
| lower_x, | |
| lower_y, | |
| "g-", | |
| linewidth=2, | |
| label="Rising Support", | |
| alpha=0.8, | |
| ) | |
| else: | |
| ax.plot( | |
| lower_x, lower_y, "g--", linewidth=2, label="Support", alpha=0.8 | |
| ) | |
| ax.legend(loc="best", fontsize=8) | |
| def close_figure(self, fig: Figure) -> None: | |
| """Close matplotlib figure to free memory.""" | |
| plt.close(fig) | |