Python实现自动检测股票高低点及虚拟趋势线突破预警
Hey there! Let's work through your problem together. First, I spot a syntax error in your current code—you're missing a function like np.where to convert those boolean conditions into 1/0 values. Even once fixed, the strict "five-finger rule" might struggle with 5-minute intraday data's inherent noise, leading to either too many false signals or missed valid levels. Let's break this down into two parts: improving high/low detection, then building the breakout alert system.
1. Fixing & Enhancing High/Low Point Detection
First, Correct the Syntax
Your current code is incomplete—here's how to fix it with numpy.where to properly assign 1s and 0s for highs/lows:
import numpy as np import pandas as pd # Assuming df is your 5-minute OHLC DataFrame df['Highs'] = np.where( (df['High'] > df['High'].shift(1)) & (df['High'] > df['High'].shift(2)) & (df['High'] > df['High'].shift(-1)) & (df['High'] > df['High'].shift(-2)), 1, 0 ) df['Lows'] = np.where( (df['Low'] < df['Low'].shift(1)) & (df['Low'] < df['Low'].shift(2)) & (df['Low'] < df['Low'].shift(-1)) & (df['Low'] < df['Low'].shift(-2)), 1, 0 )
Why the Five-Finger Rule Falls Short for Intraday Data
5-minute bars have lots of noise—strictly requiring a bar to be higher/lower than exactly two bars on either side will miss many meaningful swing highs/lows, or flag random spikes as valid levels. A better approach is using peak/trough detection with adjustable sensitivity, like scipy.signal.find_peaks:
from scipy.signal import find_peaks # Detect swing highs: adjust height/distance to filter noise high_peaks, _ = find_peaks(df['High'], height=None, distance=3) # Distance between peaks >=3 bars df['Highs'] = 0 df.loc[high_peaks, 'Highs'] = 1 # Detect swing lows (invert the Low series since find_peaks looks for peaks) low_peaks, _ = find_peaks(-df['Low'], height=None, distance=3) df['Lows'] = 0 df.loc[low_peaks, 'Lows'] = 1
distance: Ensures peaks/troughs are at least N bars apart (tweak based on your data—3-5 works well for 5-minute intraday)height: You can add a minimum price difference threshold to ignore tiny spikes (e.g.,height=0.5for a 0.5-point minimum swing)
2. Building Breakout Alerts for Support/Resistance Lines
Once you have valid highs/lows, you need to:
- Extract the key support/resistance levels (connecting consecutive valid highs/lows to form trend lines)
- Check if the latest closing price breaks above a resistance line or below a support line
Step 1: Extract Valid High/Low Points
First, get the price and index values for your detected highs and lows:
# Get swing high points (index, price) swing_highs = df[df['Highs'] == 1][['High']].reset_index() swing_highs.columns = ['bar_index', 'high_price'] # Get swing low points (index, price) swing_lows = df[df['Lows'] == 1][['Low']].reset_index() swing_lows.columns = ['bar_index', 'low_price']
Step 2: Define Support/Resistance Lines & Check Breakouts
For intraday data, we'll focus on recent trend lines (e.g., the last 2-3 swing highs/lows to form relevant resistance/support). Here's a function to check if the latest close breaks these lines:
def check_breakout(df, swing_highs, swing_lows): latest_close = df['Close'].iloc[-1] alert = None # Check resistance breakout (last two swing highs form resistance line) if len(swing_highs) >= 2: # Get last two swing highs h1_idx, h1_price = swing_highs.iloc[-2][['bar_index', 'high_price']] h2_idx, h2_price = swing_highs.iloc[-1][['bar_index', 'high_price']] # Calculate resistance line equation: y = mx + b m_resist = (h2_price - h1_price) / (h2_idx - h1_idx) b_resist = h1_price - m_resist * h1_idx # Price at latest bar index on resistance line resist_line_price = m_resist * df.index[-1] + b_resist # Check if close breaks above resistance (with a small buffer to avoid wicks) if latest_close > resist_line_price * 1.001: # 0.1% buffer alert = f"Resistance breakout detected! Close {latest_close} > Resistance {round(resist_line_price, 2)}" # Check support breakdown (last two swing lows form support line) if len(swing_lows) >= 2: l1_idx, l1_price = swing_lows.iloc[-2][['bar_index', 'low_price']] l2_idx, l2_price = swing_lows.iloc[-1][['bar_index', 'low_price']] # Calculate support line equation m_support = (l2_price - l1_price) / (l2_idx - l1_idx) b_support = l1_price - m_support * l1_idx # Price at latest bar index on support line support_line_price = m_support * df.index[-1] + b_support # Check if close breaks below support if latest_close < support_line_price * 0.999: # 0.1% buffer alert = f"Support breakdown detected! Close {latest_close} < Support {round(support_line_price, 2)}" return alert # Run the check when a new 5-minute bar is formed breakout_alert = check_breakout(df, swing_highs, swing_lows) if breakout_alert: print(breakout_alert) # Add your notification logic here (e.g., email, Telegram alert)
Key Notes for Intraday Use
- Buffer Zone: The 0.1% buffer helps avoid false breakouts from wicks—tweak this based on your asset's volatility.
- Relevant Lines: Focus on recent swing points (last 2-3) instead of all historical highs/lows, as intraday trends shift quickly.
- Noise Filtering: Combine breakout checks with volume—breakouts on higher volume are more reliable.
内容的提问来源于stack exchange,提问作者Don Coder




