公平价值缺口 (FVG) 检测与过滤：基于 ICT 概念的应用

在金融市场的技术分析中，公平价值缺口（Fair Value Gap，FVG）是由 “Inner Circle Trader” (ICT) 理论引出的重要价格形态。FVG 提供了关于市场动量和价格走势的重要信息。然而相对于使用肉眼手工盯盘，为了节省时间并提高识别 FVG 的准确性，我们开发了量化工具，该工具不仅能检测和绘制 FVG，还能通过多种过滤模式提高识别的精确度和质量。

ICT 概念简述

ICT，即 “Inner Circle Trader”，是一种交易策略，强调市场中存在一些具有影响力的大型交易者，他们会在特定的价格区间进行交易，从而在图表上留下可辨识的足迹，如公平价值缺口 (FVG)。这些智能资金会在市场中制造波动，导致价格快速移动并形成价格缺口，这些缺口在技术分析中具有重要的参考价值。

FVG 的定义

公平价值缺口 (FVG) 指的是连续三个K线不重叠的情况。根据这个定义，在上升趋势中，最后一个K线的最低价应高于第三个K线的最高价；而在下降趋势中，最后一个K线的最高价应低于第三个K线的最低价。FVG 的出现通常预示着价格的快速移动和潜在的价格反转点。

FVG 检测工具的实现

为了更高效地检测和过滤 FVG，我们开发了一个 Python 检测工具，利用 Pandas 和 NumPy 等数据处理工具，对市场数据进行分析和处理。以下是该库的主要功能和实现细节。

参数说明

• FVGFilter：决定是否应用过滤器。输入可以是 “On” 或 “Off”。
• FVGFilterType：决定应用哪种类型的过滤器。包括四种模式：“Very Defensive”、“Defensive”、“Aggressive” 和 “Very Aggressive”。
• ShowDeFVG：布尔值，决定是否显示看涨趋势中的 FVG。
• ShowSuFVG：布尔值，决定是否显示看跌趋势中的 FVG。

过滤模式的判断逻辑

1. 非过滤模式 (FVGFilter = ‘Off’)

• 看涨 FVG：当当前 K 线的最低价高于第三个 K 线的最高价时，形成看涨 FVG。
• 看跌 FVG：当当前 K 线的最高价低于第三个 K 线的最低价时，形成看跌 FVG。

2. 非常激进模式 (FVGFilter = ‘On’, FVGFilterType = ‘Very Aggressive’)

• 看涨 FVG：当当前 K 线的最低价高于第三个 K 线的最高价，并且当前 K 线的最高价高于第二个 K 线的最高价时，形成看涨 FVG。
• 看跌 FVG：当当前 K 线的最高价低于第三个 K 线的最低价，并且第二个 K 线的最低价高于当前 K 线的最低价时，形成看跌 FVG。

3. 激进模式 (FVGFilter = ‘On’, FVGFilterType = ‘Aggressive’)

• 看涨 FVG：
  • 当前 K 线的最低价高于第三个 K 线的最高价。
  • 第二个 K 线的波动范围（最高价 - 最低价）大于等于当前 K 线的平均真实波动幅度（ATR）。
  • 当前 K 线的最高价高于第二个 K 线的最高价。
• 看跌 FVG：
  • 当前 K 线的最高价低于第三个 K 线的最低价。
  • 第二个 K 线的波动范围（最高价 - 最低价）大于等于当前 K 线的 ATR。
  • 第二个 K 线的最低价高于当前 K 线的最低价。

4. 防御模式 (FVGFilter = ‘On’, FVGFilterType = ‘Defensive’)

• 看涨 FVG：
  • 当前 K 线的最低价高于第三个 K 线的最高价。
  • 第二个 K 线的波动范围（最高价 - 最低价）大于等于 1.5 倍的 ATR。
  • 当前 K 线的最高价高于第二个 K 线的最高价。
  • 第二个 K 线的最低价高于第三个 K 线的最低价。
  • 第二个 K 线和第三个 K 线都是阳线，或第二个 K 线的实体（收盘价 - 开盘价）占其波动范围的比例大于 70%。
• 看跌 FVG：
  • 当前 K 线的最高价低于第三个 K 线的最低价。
  • 第二个 K 线的波动范围（最高价 - 最低价）大于等于 1.5 倍的 ATR。
  • 第二个 K 线的最低价高于当前 K 线的最低价。
  • 第二个 K 线的最高价低于第三个 K 线的最高价。
  • 第二个 K 线和第三个 K 线都是阴线，或第二个 K 线的实体（收盘价 - 开盘价）占其波动范围的比例大于 70%。

5. 非常防御模式 (FVGFilter = ‘On’, FVGFilterType = ‘Very Defensive’)

• 看涨 FVG：
  • 当前 K 线的最低价高于第三个 K 线的最高价。
  • 第二个 K 线的波动范围（最高价 - 最低价）大于等于 1.5 倍的 ATR。
  • 当前 K 线的最高价高于第二个 K 线的最高价。
  • 第二个 K 线的最低价高于第三个 K 线的最低价。
  • 第二个 K 线和第三个 K 线都是阳线，并且第二个 K 线的实体占其波动范围的比例大于 70%。
  • 第三个 K 线的实体占其波动范围的比例大于 35%。
  • 当前 K 线的实体占其波动范围的比例大于 35%。
• 看跌 FVG：
  • 当前 K 线的最高价低于第三个 K 线的最低价。
  • 第二个 K 线的波动范围（最高价 - 最低价）大于等于 1.5 倍的 ATR。
  • 第二个 K 线的最低价高于当前 K 线的最低价。
  • 第二个 K 线的最高价低于第三个 K 线的最高价。
  • 第二个 K 线和第三个 K 线都是阴线，并且第二个 K 线的实体占其波动范围的比例大于 70%。
  • 第三个 K 线的实体占其波动范围的比例大于 35%。
  • 当前 K 线的实体占其波动范围的比例大于 35%。

通过以上详细的过滤模式判断逻辑，交易者可以根据自身的风险偏好和市场判断，灵活应用不同的过滤模式，从而更精准地识别和利用市场中的公平价值缺口 (FVG)。

代码实现

import pandas as pd
import numpy as np

def FVGDetector(data, FVGFilter='On', FVGFilterType='Defensive', ShowDeFVG=True, ShowSuFVG=True):
    # 计算ATR
    data['ATR'] = data['Close'].rolling(window=6, min_periods=1).apply(lambda x: pd.Series(x).diff().abs().mean(), raw=False)
    
    def check_DConditionFVG(row, prev_row, prev2_row):
        if FVGFilter == 'Off':
            return row['Low'] > prev2_row['High']
        elif FVGFilter == 'On':
            if FVGFilterType == 'Very Aggressive':
                return row['Low'] > prev2_row['High'] and row['High'] > prev_row['High']
            elif FVGFilterType == 'Aggressive':
                return (row['Low'] > prev2_row['High'] and (prev_row['High'] - prev_row['Low']) >= (1.0 * row['ATR']) and row['High'] > prev_row['High'])
            elif FVGFilterType == 'Defensive':
                return (row['Low'] > prev2_row['High'] and (prev_row['High'] - prev_row['Low']) >= (1.5 * row['ATR']) and row['High'] > prev_row['High'] and prev_row['Low'] > prev2_row['Low'] and
                        ((prev2_row['Close'] - prev2_row['Open'] > 0 and prev_row['Close'] - prev_row['Open'] > 0) or abs((prev_row['Close'] - prev_row['Open']) / (prev_row['High'] - prev_row['Low'])) > 0.7))
            elif FVGFilterType == 'Very Defensive':
                return (row['Low'] > prev2_row['High'] and (prev_row['High'] - prev_row['Low']) >= (1.5 * row['ATR']) and row['High'] > prev_row['High'] and prev_row['Low'] > prev2_row['Low'] and
                        ((prev2_row['Close'] - prev2_row['Open'] > 0 and prev_row['Close'] - prev_row['Open'] > 0) and abs((prev_row['Close'] - prev_row['Open']) / (prev_row['High'] - prev_row['Low'])) > 0.7) and
                        abs((prev2_row['Close'] - prev2_row['Open']) / (prev2_row['High'] - prev2_row['Low'])) > 0.35 and abs((row['Close'] - row['Open']) / (row['High'] - row['Low'])) > 0.35)
        return False
    
    def check_SConditionFVG(row, prev_row, prev2_row):
        if FVGFilter == 'Off':
            return row['High'] < prev2_row['Low']
        elif FVGFilter == 'On':
            if FVGFilterType == 'Very Aggressive':
                return row['High'] < prev2_row['Low'] and prev_row['Low'] > row['Low']
            elif FVGFilterType == 'Aggressive':
                return (row['High'] < prev2_row['Low'] and (prev_row['High'] - prev_row['Low']) >= (1.0 * row['ATR']) and prev_row['Low'] > row['Low'])
            elif FVGFilterType == 'Defensive':
                return (row['High'] < prev2_row['Low'] and (prev_row['High'] - prev_row['Low']) >= (1.5 * row['ATR']) and prev_row['Low'] > row['Low'] and prev_row['High'] < prev2_row['High'] and
                        ((prev2_row['Close'] - prev2_row['Open'] < 0 and prev_row['Close'] - prev_row['Open'] < 0) or abs((prev_row['Close'] - prev_row['Open']) / (prev_row['High'] - prev_row['Low'])) > 0.7))
            elif FVGFilterType == 'Very Defensive':
                return (row['High'] < prev2_row['Low'] and (prev_row['High'] - prev_row['Low']) >= (1.5 * row['ATR']) and prev_row['Low'] > row['Low'] and prev_row['High'] < prev2_row['High'] and
                        ((prev2_row['Close'] - prev2_row['Open'] < 0 and prev_row['Close'] - prev_row['Open'] < 0) and abs((prev_row['Close'] - prev_row['Open']) / (prev_row['High'] - prev_row['Low'])) > 0.7) and
                        abs((prev2_row['Close'] - prev2_row['Open']) / (prev2_row['High'] - prev2_row['Low'])) > 0.35 and abs((row['Close'] - row['Open']) / (row['High'] - row['Low'])) > 0.35)
        return False
    
    data['DConditionFVG'] = data.apply(lambda row: check_DConditionFVG(
    data.loc[row.name-1] if row.name > 0 else data.loc[row.name],
    data.loc[row.name-2] if row.name > 1 else data.loc[row.name],
    data.loc[row.name-3] if row.name > 2 else data.loc[row.name]
    ), axis=1)

    data['SConditionFVG'] = data.apply(lambda row: check_SConditionFVG(
        data.loc[row.name-1] if row.name > 0 else data.loc[row.name],
        data.loc[row.name-2] if row.name > 1 else data.loc[row.name],
        data.loc[row.name-3] if row.name > 2 else data.loc[row.name]
    ), axis=1)

    def find_FVG_zones(data):
        data['DDFVG'] = np.nan
        data['DPFVG'] = np.nan
        data['SDFVG'] = np.nan
        data['SPFVG'] = np.nan
        
        for index, row in data.iterrows():
            if row['DConditionFVG']:
                data.at[index, 'DDFVG'] = row['High']
                data.at[index, 'DPFVG'] = row['Low']
            if row['SConditionFVG']:
                data.at[index, 'SDFVG'] = row['Low']
                data.at[index, 'SPFVG'] = row['High']
                    
        return data
    
    data = find_FVG_zones(data)
    
    return data

def main():
    while True:
        exchange.SetContractType('ag2408')
        r = exchange.GetRecords()
        selected_data = [{key: item[key] for key in ['Time', 'Open', 'High', 'Low', 'Close']} for item in r]

        # 创建DataFrame
        df = pd.DataFrame(selected_data)

        # 运行FVG检测器
        result = FVGDetector(df, FVGFilter='On', FVGFilterType='Defensive', ShowDeFVG=True, ShowSuFVG=True)

        # 转换日期格式
        result['date'] = df['Time']

        ext.PlotRecords(r, "KLine")

        row = result.iloc[len(result)-1]

        if not np.isnan(row['DDFVG']) and not np.isnan(row['DPFVG']):
            Log(_D(row["Time"]/1000), 'DDFVG出现')
            ext.PlotFlag(int(row["Time"]), "DDFVG", "DDFVG")
        if not np.isnan(row['SDFVG']) and not np.isnan(row['SPFVG']):
            Log(_D(row["Time"]/1000), 'SDFVG出现')
            ext.PlotFlag(int(row["Time"]), "SDFVG", "SDFVG")

        Sleep(1000 * 60 * 10)

结果展示与分析

我们举例示范一下，以ag2408沪银合约10分钟的数据举例示范一下，可以看到当满足相应条件时，会返回相应的’SDFVG’和’DDFVG’指标，证明测试代码运行无误，这里我们也可以更换四种不同的模式，用来符合不同的交易标准。

该检测库能够实时识别并绘制市场中的 FVG，并且可以根据用户设定的过滤模式，对不同类型的 FVG 进行过滤和显示。通过对 ATR 和价格走势的结合分析，可以更准确地识别市场中的有效 FVG，从而为交易决策提供支持。

通过此库，我们可以更高效地识别和利用市场中的 FVG，从而在交易中获得更大的优势。无论是用于趋势交易还是逆向交易，FVG 都可以提供有价值的市场信号，帮助大家更好地把握市场动向。