如何自动化检测公允价值缺口（FVG）实现更智能的交易

pip install pandas numpy yfinance mplfinance ccxt

import pandas as pd
import numpy as np
import yfinance as yf
import mplfinance as mpf
from datetime import datetime, timedelta

def fetch_ohlcv(symbol: str, interval: str = '1h', days: int = 30) -> pd.DataFrame:
    """
    从Yahoo Finance获取OHLCV数据。
    加密货币请使用ccxt替代。
    """
    end = datetime.now()
    start = end - timedelta(days=days)

    ticker = yf.Ticker(symbol)
    df = ticker.history(start=start, end=end, interval=interval)

    df.columns = [c.lower() for c in df.columns]
    df = df[['open', 'high', 'low', 'close', 'volume']]
    df.index.name = 'datetime'

    return df

from dataclasses import dataclass
from typing import List, Optional
from enum import Enum

class FVGType(Enum):
    BULLISH = "bullish"
    BEARISH = "bearish"

@dataclass
class FVG:
    """表示一个公允价值缺口"""
    type: FVGType
    top: float
    bottom: float
    formed_at: datetime
    formed_idx: int
    mitigated: bool = False
    mitigated_at: Optional[datetime] = None
    mitigated_pct: float = 0.0

    @property
    def size(self) -> float:
        return self.top - self.bottom

    @property
    def midpoint(self) -> float:
        return (self.top + self.bottom) / 2

def detect_fvg(df: pd.DataFrame, min_gap_pct: float = 0.001) -> List[FVG]:
    """
    在OHLCV数据中检测公允价值缺口。

    参数:
        df: 包含open, high, low, close列的DataFrame
        min_gap_pct: 缺口最小百分比（过滤噪音）

    返回:
        FVG对象列表
    """
    fvgs = []

    for i in range(2, len(df)):
        candle_1 = df.iloc[i - 2]
        candle_2 = df.iloc[i - 1]  # 检测中不使用，但保留
        candle_3 = df.iloc[i]

        current_price = candle_3['close']
        min_gap_size = current_price * min_gap_pct

        # 看涨FVG: K线3最低价 > K线1最高价
        if candle_3['low'] > candle_1['high']:
            gap_size = candle_3['low'] - candle_1['high']
            if gap_size >= min_gap_size:
                fvg = FVG(
                    type=FVGType.BULLISH,
                    top=candle_3['low'],
                    bottom=candle_1['high'],
                    formed_at=df.index[i],
                    formed_idx=i
                )
                fvgs.append(fvg)

        # 看跌FVG: K线3最高价 < K线1最低价
        elif candle_3['high'] < candle_1['low']:
            gap_size = candle_1['low'] - candle_3['high']
            if gap_size >= min_gap_size:
                fvg = FVG(
                    type=FVGType.BEARISH,
                    top=candle_1['low'],
                    bottom=candle_3['high'],
                    formed_at=df.index[i],
                    formed_idx=i
                )
                fvgs.append(fvg)

    return fvgs

def track_mitigation(df: pd.DataFrame, fvgs: List[FVG],
                     lookforward: int = 50) -> List[FVG]:
    """
    追踪FVG是否被缓解（价格返回填补缺口）。

    参数:
        df: FVG形成后继续的价格数据
        fvgs: 检测到的FVG列表
        lookforward: 检查缓解的K线数量

    返回:
        带有缓解数据的更新FVG列表
    """
    for fvg in fvgs:
        if fvg.formed_idx >= len(df) - 1:
            continue

        # 检查后续K线是否缓解
        end_idx = min(fvg.formed_idx + lookforward, len(df))

        for i in range(fvg.formed_idx + 1, end_idx):
            candle = df.iloc[i]

            if fvg.type == FVGType.BULLISH:
                # 看涨FVG在价格跌入缺口时被缓解
                if candle['low'] <= fvg.top:
                    fvg.mitigated = True
                    fvg.mitigated_at = df.index[i]

                    # 计算缺口被填补了多少
                    lowest_point = candle['low']
                    if lowest_point <= fvg.bottom:
                        fvg.mitigated_pct = 1.0  # 完全填补
                    else:
                        filled = fvg.top - lowest_point
                        fvg.mitigated_pct = filled / fvg.size
                    break

            else:  # 看跌FVG
                # 看跌FVG在价格涨入缺口时被缓解
                if candle['high'] >= fvg.bottom:
                    fvg.mitigated = True
                    fvg.mitigated_at = df.index[i]

                    highest_point = candle['high']
                    if highest_point >= fvg.top:
                        fvg.mitigated_pct = 1.0
                    else:
                        filled = highest_point - fvg.bottom
                        fvg.mitigated_pct = filled / fvg.size
                    break

    return fvgs

def calculate_fvg_statistics(fvgs: List[FVG]) -> dict:
    """生成FVG行为的统计数据。"""
    if not fvgs:
        return {}

    bullish = [f for f in fvgs if f.type == FVGType.BULLISH]
    bearish = [f for f in fvgs if f.type == FVGType.BEARISH]

    mitigated = [f for f in fvgs if f.mitigated]

    stats = {
        'total_fvgs': len(fvgs),
        'bullish_count': len(bullish),
        'bearish_count': len(bearish),
        'mitigated_count': len(mitigated),
        'mitigation_rate': len(mitigated) / len(fvgs) if fvgs else 0,
        'avg_fill_pct': np.mean([f.mitigated_pct for f in mitigated]) if mitigated else 0,
        'avg_gap_size': np.mean([f.size for f in fvgs])
    }

    return stats

def main():
    # 获取数据 - 以SPY为例
    print("获取数据中...")
    df = fetch_ohlcv("SPY", interval="1h", days=60)
    print(f"加载了 {len(df)} 根K线")

    # 检测FVG
    print("\n检测公允价值缺口中...")
    fvgs = detect_fvg(df, min_gap_pct=0.001)  # 0.1%最小缺口
    print(f"发现 {len(fvgs)} 个FVG")

    # 追踪缓解
    print("\n追踪缓解中...")
    fvgs = track_mitigation(df, fvgs, lookforward=50)

    # 计算统计
    stats = calculate_fvg_statistics(fvgs)

    print("\n" + "="*50)
    print("FVG统计数据")
    print("="*50)
    print(f"检测到的FVG总数: {stats['total_fvgs']}")
    print(f"  - 看涨: {stats['bullish_count']}")
    print(f"  - 看跌: {stats['bearish_count']}")
    print(f"缓解率: {stats['mitigation_rate']*100:.1f}%")
    print(f"缓解时的平均填补: {stats['avg_fill_pct']*100:.1f}%")
    print(f"平均缺口大小: ${stats['avg_gap_size']:.2f}")

if __name__ == "__main__":
    main()

加载了 720 根K线

检测公允价值缺口中...
发现 47 个FVG

追踪缓解中...

==================================================
FVG统计数据
==================================================
检测到的FVG总数: 47
  - 看涨: 28
  - 看跌: 19
缓解率: 72.3%
缓解时的平均填补: 84.2%
平均缺口大小: $0.43

import ccxt

def fetch_crypto_ohlcv(symbol: str, exchange: str = 'binance',
                       timeframe: str = '1h', limit: int = 500) -> pd.DataFrame:
    """
    从加密货币交易所获取OHLCV数据。
    """
    exchange_class = getattr(ccxt, exchange)
    ex = exchange_class()

    ohlcv = ex.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)

    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
    df.set_index('datetime', inplace=True)
    df.drop('timestamp', axis=1, inplace=True)

    return df

# 使用方法
df_btc = fetch_crypto_ohlcv("BTC/USDT", exchange="binance", timeframe="1h")
fvgs_btc = detect_fvg(df_btc, min_gap_pct=0.002)  # 加密货币波动性更高所以阈值更高