Implement ZIGZAG

[febuz]

Zig Zag: filters out price moves smaller than pct percent and connects the remaining swing highs and lows with straight lines.

Theory: ignoring countermoves below the threshold exposes the underlying wave structure for Elliott Wave counts, chart patterns and swing measurement. The last leg is provisional and repaints until confirmed, so it describes the past and is not a realtime signal.
Returns: list of floats
Source: https://chartschool.stockcharts.com/table-of-contents/technical-indicators-and-overlays/technical-overlays/zigzag

Review notes

• Single-indicator change, one file, one commit.
• Pure Python; smoke-tested on synthetic OHLCV with warm-up assertions.

[joosthoeks]

Maybe more suitable in event_driven.

[febuz]

Is possible, but then it is an implementation choice
Gemini: class ZigZagIndicator:
"""
An event-driven implementation of the Zig Zag indicator.

Filters out price movements smaller than 'pct' percent and connects
the remaining swing highs and lows with straight lines.

Because Zig Zag is inherently retrospective (the last leg is provisional
and "repaints" until a reversal is confirmed), this class maintains
the full history of confirmed and provisional points.
"""
def __init__(self, pct: float):
    if pct <= 0:
        raise ValueError("Percentage threshold 'pct' must be greater than 0.")
    
    self.threshold = pct / 100.0
    
    # Series tracking
    self.prices = []  # Stores the input sequence (e.g., Close prices)
    
    # States: 
    #  1: Currently searching for a Peak (current direction is UP)
    # -1: Currently searching for a Valley (current direction is DOWN)
    #  0: Uninitialized (waiting for initial direction)
    self.state = 0
    
    # Indices and values of the last confirmed and extreme points
    self.last_extreme_idx = 0
    self.last_extreme_val = None
    
    # List of confirmed anchor points: tuples of (index, price)
    self.anchors = [] 

def update(self, price: float) -> list[float]:
    """
    Feed a new price point into the indicator.
    Returns the full historical Zig Zag series (reconstructed) up to the current step.
    """
    self.prices.append(price)
    current_idx = len(self.prices) - 1
    
    # 1. Initialization Step
    if len(self.prices) < 2:
        self.last_extreme_idx = 0
        self.last_extreme_val = price
        self.anchors = [(0, price)]
        return [price]

    if self.state == 0:
        # Determine initial direction based on the first price movement
        initial_price = self.prices[0]
        if price > initial_price:
            # Price moved up; we are now looking for a Peak.
            # The first point is treated as an initial Valley.
            self.state = 1  
            self.last_extreme_idx = current_idx
            self.last_extreme_val = price
        elif price < initial_price:
            # Price moved down; we are now looking for a Valley.
            # The first point is treated as an initial Peak.
            self.state = -1  
            self.last_extreme_idx = current_idx
            self.last_extreme_val = price
        else:
            # Price unchanged, state remains uninitialized
            pass
        
        return self._interpolate_zigzag()

    # 2. State Machine Update
    if self.state == 1:
        # Looking for a Peak
        if price > self.last_extreme_val:
            # We found a higher high; update our provisional peak candidate
            self.last_extreme_val = price
            self.last_extreme_idx = current_idx
        else:
            # Check for reversal: has the price dropped below the threshold from the provisional peak?
            reversal_threshold = self.last_extreme_val * (1 - self.threshold)
            if price <= reversal_threshold:
                # Reversal confirmed! 
                # 1. Lock in the peak we found as a confirmed anchor.
                self.anchors.append((self.last_extreme_idx, self.last_extreme_val))
                # 2. Shift state to looking for a Valley (-1).
                self.state = -1
                # 3. Initialize the new Valley candidate to the current price.
                self.last_extreme_val = price
                self.last_extreme_idx = current_idx

    elif self.state == -1:
        # Looking for a Valley
        if price < self.last_extreme_val:
            # We found a lower low; update our provisional valley candidate
            self.last_extreme_val = price
            self.last_extreme_idx = current_idx
        else:
            # Check for reversal: has the price risen above the threshold from the provisional valley?
            reversal_threshold = self.last_extreme_val * (1 + self.threshold)
            if price >= reversal_threshold:
                # Reversal confirmed!
                # 1. Lock in the valley we found as a confirmed anchor.
                self.anchors.append((self.last_extreme_idx, self.last_extreme_val))
                # 2. Shift state to looking for a Peak (1).
                self.state = 1
                # 3. Initialize the new Peak candidate to the current price.
                self.last_extreme_val = price
                self.last_extreme_idx = current_idx

    return self._interpolate_zigzag()

def _interpolate_zigzag(self) -> list[float]:
    """
    Constructs the straight-line Zig Zag series by interpolating between
    all confirmed anchors and the current provisional extreme / end point.
    """
    n = len(self.prices)
    zigzag_series = [0.0] * n
    
    # Build the final list of vertices to interpolate.
    # This includes all confirmed anchors, the last known provisional extreme,
    # and terminates at the current actual price (index n-1).
    vertices = list(self.anchors)
    
    # Append the provisional extreme if it's not already the latest anchor
    if not vertices or vertices[-1][0] != self.last_extreme_idx:
        vertices.append((self.last_extreme_idx, self.last_extreme_val))
        
    # Append the very last raw price point to tie off the series
    if vertices[-1][0] != n - 1:
        vertices.append((n - 1, self.prices[-1]))
        
    # Interpolate between consecutive vertices
    for i in range(len(vertices) - 1):
        idx1, val1 = vertices[i]
        idx2, val2 = vertices[i+1]
        
        if idx1 == idx2:
            zigzag_series[idx1] = val2
            continue
            
        # Linear interpolation formula
        for t in range(idx1, idx2 + 1):
            fraction = (t - idx1) / (idx2 - idx1)
            zigzag_series[t] = val1 + fraction * (val2 - val1)
            
    return zigzag_series

This is a usage decision, not a package decision, logically it resides with pattern, as between high and low, you make the simples pattern, a trendline, automatically.

[febuz]

So I would allow for an event.py in which you allow indicators to be used in more advanced defs from other .py files, plus I corrected the broken link in the initial pull request