在金融市场的技术分析中，公平价值缺口（Fair Value Gap，FVG）是由 “Inner Circle Trader” (ICT) 理论引出的重要价格形态。FVG 提供了关于市场动量和价格走势的重要信息。然而相对于使用肉眼手工盯盘，为了节省时间并提高识别 FVG 的准确性，我们开发了量化工具，该工具不仅能检测和绘制 FVG，还能通过多种过滤模式提高识别的精确度和质量。
ICT，即 “Inner Circle Trader”，是一种交易策略，强调市场中存在一些具有影响力的大型交易者，他们会在特定的价格区间进行交易，从而在图表上留下可辨识的足迹，如公平价值缺口 (FVG)。这些智能资金会在市场中制造波动，导致价格快速移动并形成价格缺口，这些缺口在技术分析中具有重要的参考价值。
公平价值缺口 (FVG) 指的是连续三个K线不重叠的情况。根据这个定义，在上升趋势中，最后一个K线的最低价应高于第三个K线的最高价；而在下降趋势中，最后一个K线的最高价应低于第三个K线的最低价。FVG 的出现通常预示着价格的快速移动和潜在的价格反转点。
为了更高效地检测和过滤 FVG，我们开发了一个 Python 检测工具，利用 Pandas 和 NumPy 等数据处理工具，对市场数据进行分析和处理。以下是该库的主要功能和实现细节。
1. 非过滤模式 (FVGFilter = 'Off')
2. 非常激进模式 (FVGFilter = 'On', FVGFilterType = 'Very Aggressive')
3. 激进模式 (FVGFilter = 'On', FVGFilterType = 'Aggressive')
4. 防御模式 (FVGFilter = 'On', FVGFilterType = 'Defensive')
5. 非常防御模式 (FVGFilter = 'On', FVGFilterType = 'Very Defensive')
通过以上详细的过滤模式判断逻辑，交易者可以根据自身的风险偏好和市场判断，灵活应用不同的过滤模式，从而更精准地识别和利用市场中的公平价值缺口 (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 都可以提供有价值的市场信号，帮助大家更好地把握市场动向。
转载自：优宽量化交易平台社区
作者：ianzeng123
