def label_swing_waves(self, swings_df: pd.DataFrame) -> List[Dict]: """ Convert alternating swing points into wave segments. Returns list of waves with direction, length, and ratio info. """ if len(swings_df) < 2: return []
impulse_ok = self.check_impulse_rules(waves) corrective_ok = self.check_corrective_rules(waves)
A, B, C = waves[:3] # Typical rule: B retraces 0.382 to 0.886 of A retrace_ratio = B['magnitude'] / A['magnitude'] if A['magnitude'] != 0 else 0 if 0.382 <= retrace_ratio <= 0.886: # C often equals A in length (1.0 or 1.618) c_ratio = C['magnitude'] / A['magnitude'] if 0.618 <= c_ratio <= 1.618: return True return False
swings = sorted(swings, key=lambda x: x['index']) return pd.DataFrame(swings) elliott wave python code
# Rule 3: Wave 4 price overlap with Wave 1? # For uptrend impulse: w1 up, w2 down, w3 up, w4 down, w5 up # Overlap means low of w4 < high of w1 if w1['direction'] == 'up': wave1_high = max(w1['start_price'], w1['end_price']) wave4_low = min(w4['start_price'], w4['end_price']) if wave4_low <= wave1_high: return False else: # downtrend impulse wave1_low = min(w1['start_price'], w1['end_price']) wave4_high = max(w4['start_price'], w4['end_price']) if wave4_high >= wave1_low: return False
return True
if len(waves) < 5: return {'pattern': 'none', 'waves': waves, 'valid': False, 'reason': 'Not enough swing points'} def label_swing_waves(self, swings_df: pd
# Plotting plt.figure(figsize=(14, 6)) plt.plot(price_series, label='Price', color='black', alpha=0.6)
def detect_elliott_waves(self, prices: np.ndarray) -> Dict: """ Main function: returns detected wave structure and validation. """ swings_df = self.find_swing_points(prices) waves = self.label_swing_waves(swings_df)
# Rule 2: Wave 3 not shortest if w3['magnitude'] <= w1['magnitude'] or w3['magnitude'] <= w5['magnitude']: if w3['magnitude'] < w1['magnitude'] and w3['magnitude'] < w5['magnitude']: return False # For uptrend impulse: w1 up, w2 down,
def check_corrective_rules(self, waves: List[Dict]) -> bool: """Check 3-wave corrective pattern (A,B,C).""" if len(waves) < 3: return False
def fibonacci_ratios(self, wave: Dict) -> Dict: """Calculate Fibonacci retracements/extensions for a wave.""" mag = wave['magnitude'] return { '0.382': mag * 0.382, '0.5': mag * 0.5, '0.618': mag * 0.618, '1.0': mag, '1.272': mag * 1.272, '1.618': mag * 1.618, }
# Rule 1: Wave 2 retrace < 100% of Wave 1 if w2['magnitude'] >= w1['magnitude']: return False
# Annotate wave numbers (first 5 waves if exist) waves = result['waves'] for i, wave in enumerate(waves[:5]): mid_idx = (wave['start_idx'] + wave['end_idx']) // 2 mid_price = (wave['start_price'] + wave['end_price']) / 2 plt.text(mid_idx, mid_price, str(i+1), fontsize=12, fontweight='bold', bbox=dict(facecolor='yellow', alpha=0.7))
def check_impulse_rules(self, waves: List[Dict]) -> bool: """ Validate Elliott Wave impulse pattern (5 waves: 1,2,3,4,5). Rules: 1. Wave 2 cannot retrace more than 100% of Wave 1. 2. Wave 3 is never the shortest (in magnitude). 3. Wave 4 does not overlap Wave 1 (in price). 4. Wave 3 often shows 1.618 extension of Wave 1. """ if len(waves) < 5: return False