输入/搜索内容
ianzeng123
1
关注
183
关注者
公平价值缺口 (FVG) 检测与过滤：基于 ICT 概念的应用
原创教程
创建于 2024-07-11 17:59:47  更新于 2024-11-19 13:27:13
 0
 1622

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)。

代码实现 


            python
            
            
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 都可以提供有价值的市场信号，帮助大家更好地把握市场动向。

相关推荐
商品期货「订单流」系列文章（一）：订单流概念介绍浅谈枢纽点理论在商品期货中的应用利弗莫尔买入法原理及其在商品期货市场的应用商品期货中‘盈利加仓，亏损减仓’策略示例点石为金：如何让一个指标“化腐朽为神奇”嗤之以鼻 or 逐浪学习：浅谈波浪理论在商品期货市场中的应用策略界面参数与交互控件新增功能详解浅谈期权交易策略在不同市场行情中的应用浅谈商品期权市场中的隐含波动率微笑曲线浅谈商品期权市场平仓操作盈亏估算方法
评论
全部评论 (0)
暂无数据
暂无数据
  • 1
  • 20 条/页
iPhone 下载
Android 下载
社区
回测系统
API文档
关于我们
产品功能
创建实盘
新建策略
© 2015 - ∞ YouQuant 豫ICP备19046564号