You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

Python实现自动检测股票高低点及虚拟趋势线突破预警

Optimizing OHLC High/Low Detection & Breakout Alerts for 5-Minute Intraday Data

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.5 for a 0.5-point minimum swing)

2. Building Breakout Alerts for Support/Resistance Lines

Once you have valid highs/lows, you need to:

  1. Extract the key support/resistance levels (connecting consecutive valid highs/lows to form trend lines)
  2. 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

火山引擎 最新活动