资源加载中... loading...

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

Author: ianzeng123, Created: 2024-07-11 17:59:47, Updated: 2024-07-15 21:21:56

img

在金融市场的技术分析中,公平价值缺口(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’指标,证明测试代码运行无误,这里我们也可以更换四种不同的模式,用来符合不同的交易标准。

img

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

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


更多内容