如何自动化检测公允价值缺口（FVG）实现更智能的交易
厌倦了在图表上手动画FVG水平线，以及在交易群里反复回答同样的问题？这篇教程教你如何用Python自动化整个过程，从数据获取到可视化和回测。
上周我看到一位交易者的推文，完美地诠释了为什么我要写这篇教程。他抱怨自己"太忙了，一直在回答同样的问题"。什么样的问题呢？比如：这是个指标吗？你从哪里获取数据？怎么让它画出水平线？你是手动画的吗？
讽刺的是，这位交易者专门做期货交易的公允价值缺口（FVG）统计分析。他显然有技术背景。但他仍然在手动识别和标注这些缺口。同样的缺口，遵循算法模式。同样的缺口，脚本可以在毫秒内检测出来。
如果你花几个小时标注图表，或者更糟糕的是，为一些做得一般的指标付费，这篇教程会改变你对FVG交易的方式。我们将用Python构建一个完整的FVG检测系统。不是玩具示例。是一个真实的系统，可以获取实时数据、识别缺口、追踪其生命周期，并清晰地可视化一切。
读完之后，你将拥有可以针对任何市场运行的代码。更重要的是，你会深刻理解其中的逻辑，足以根据你的特定交易风格进行修改。
公允价值缺口形成于价格移动过于激进，以至于在图表上留下不平衡的情况。在Smart Money Concepts术语中，这代表市场经常会回来"填补"的低效价格行为。
机械定义很直接。对于看涨FVG，你需要三根连续的K线，其中第三根K线的最低价高于第一根K线的最高价。它们之间的缺口就是FVG。看跌FVG是镜像的。
使FVG可交易的不仅仅是它们的形成。而是之后发生的事情。价格倾向于重新访问这些区域。有时完全填补，有时部分填补。追踪这种行为正是自动化变得无价的地方。
大多数教程止步于检测。我们要走得更远。我们将追踪缺口的缓解、计算填补百分比，并为回测FVG策略建立基础。
你需要一些库。Pandas用于数据操作，yfinance用于免费市场数据（如果你交易加密货币则用ccxt），matplotlib加mplfinance用于可视化。
pip install pandas numpy yfinance mplfinance ccxt
数据源的选择很重要。对于股票和期货，yfinance对于原型设计来说很好用。对于加密货币，ccxt让你可以访问几十个交易所。对于生产系统，你需要一个正规的数据供应商，但免费来源对于开发来说足够了。
import pandas as pd
import numpy as np
import yfinance as yf
import mplfinance as mpf
from datetime import datetime, timedelta
魔法发生在这里。检测逻辑以三根K线为一组扫描，检查第一根和第三根K线之间是否存在第二根K线没有填补的缺口。
from dataclasses import dataclass
from typing import List, Optional
from enum import Enum
class FVGType(Enum):
BULLISH = "bullish"
min_gap_pct参数至关重要。没有它，你会检测到成千上万个没有交易意义的微小缺口。从0.1%（0.001）开始，根据资产的波动性进行调整。
检测只是故事的一半。真正的价值来自于追踪这些缺口随时间的变化。它们被填补了吗？多快？这些问题对策略开发很重要。
def track_mitigation(df: pd.DataFrame, fvgs: List[FVG],
lookforward: int = 50) -> List[FVG]:
"""
追踪FVG是否被缓解（价格返回填补缺口）。
参数:
df: FVG形成后继续的价格数据
fvgs: 检测到的FVG列表
lookforward: 检查缓解的K线数量
缓解追踪揭示了手动分析会错过的模式。当你对历史数据运行这个时，你开始看到诸如"这个资产上75%的看涨FVG在20根K线内至少被部分填补"这样的结论。这就是可操作的情报。
这是一个把所有东西结合在一起的完整示例。这个脚本获取数据、检测FVG、追踪缓解，并输出统计数据。
def main():
# 获取数据 - 以SPY为例
print("获取数据中...")
df = fetch_ohlcv("SPY", interval="1h", days=60)
当你对SPY小时数据运行这个时，你会看到类似这样的输出：
加载了 720 根K线
检测公允价值缺口中...
发现 47 个FVG
追踪缓解中...
==================================================
FVG统计数据
==================================================
检测到的FVG总数: 47
- 看涨: 28
- 看跌: 19
缓解率: 72.3%
缓解时的平均填补: 84.2%
平均缺口大小: $0.43
这些数字讲述了一个故事。72%的缓解率意味着大多数缺口最终确实会被填补。84%的平均填补表明价格通常会深入回撤到缺口区域。这些是塑造交易策略的洞察。
同样的逻辑适用于加密货币。只需将数据源从yfinance换成ccxt。
import ccxt
def fetch_crypto_ohlcv(symbol: str, exchange: str = 'binance',
timeframe: str = '1h', limit: int = 500) -> pd.DataFrame:
"""
加密市场波动性更大，所以你可能想把min_gap_pct提高到0.2%或更高。否则你会被噪音淹没。
大多数FVG系统的不足之处在这里。它们检测缺口但不区分高概率和低概率设置。并非所有FVG都是平等的。
影响FVG质量的因素包括趋势背景（顺势FVG更强）、相对于近期波动性的缺口大小、期货的时间段、以及缺口形成期间的成交量。
为FVG质量建立评分系统是你开始发展真正优势的地方。这也是像EKX.AI扫描器这样的工具变得有价值的地方。虽然上面的代码处理检测，但识别哪些FVG值得交易需要分析更广泛的市场背景。聪明钱的移动、异常的成交量模式和流动性条件都是因素。
扫描器通过呈现使某些缺口更重要的市场条件来补充手动FVG分析。当你看到一个FVG正好在大户钱包开始积累时形成，那是与随机噪音中出现的FVG不同的信号。
自动检测是第一步。真正的优势来自于将技术形态与链上数据和成交量分析结合。聚合多种信号类型的工具有助于从真正的机会中过滤噪音。
在几十种资产上运行FVG检测后，出现了哪些问题的模式。
过度检测发生在你的缺口阈值太小的时候。你最终会得到数百个没有统计意义的微小缺口。从更大的阈值开始，然后向下调整。
忽略背景导致逆势交易FVG。强势下跌趋势中的看涨FVG通常只是被碾压。在采取信号之前添加趋势过滤器。
不追踪结果意味着你永远不知道你的FVG策略是否真的有效。从第一天就内置日志记录。追踪每个检测到的缺口、每次填补、每个结果。
曲线拟合发生在你优化历史数据上的参数直到它们看起来完美的时候。然后它们在新数据上失败。始终使用样本外测试。
你现在有了一个完整的FVG检测和追踪系统。代码足够模块化，可以向几个方向扩展。
对于交易机器人，添加订单执行逻辑和仓位管理。对于扫描器产品，用web API包装它并构建前端。对于研究，对多年的数据运行它并发布你的发现。
持续从FVG获利的交易者不是那些拥有最花哨指标的人。他们是那些深刻理解底层机制、细致追踪结果、并根据数据不断完善方法的人。
这篇教程给你提供了基础。你在此基础上构建什么才是真正工作开始的地方。
从一种资产开始。运行检测一个月。追踪发生了什么。然后扩展。最好的交易系统是逐步构建的，每一层在添加下一层之前都经过测试。
代码是你的，可以修改、扩展和部署。公允价值缺口只是Smart Money Concepts拼图的一块。但这一块，如果自动化得当，每周可以节省数小时的手动图表工作。
节省的时间会复利。用它在其他地方寻找优势，或者只是以更少的压力交易。无论哪种方式，你现在都有能力做大多数零售交易者做不到的事情：让代码做繁琐的工作，而你专注于重要的决策。
邮件列表
加入我们的社区
订阅邮件列表，及时获取最新消息和更新
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
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
返回:
带有缓解数据的更新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
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()
从加密货币交易所获取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) # 加密货币波动性更高所以阈值更高
如何自动化检测公允价值缺口（FVG）实现更智能的交易 | EKX - 加密货币预涨扫描器
