Bybit自动化供需区扫描器:Telegram实时交易信号
基于Python的系统扫描Bybit所有USDT交易对,运用聪明钱概念识别供需区,通过pinbar形态追踪价格回测,并即时发送带图表标记的交易信号至Telegram。处理数百种资产仅需数分钟,替代了数小时的手动分析。
逻辑遵循链条:区域检测 → 价格回测 → 价格行为反应 → 信号生成。区域通过ATR、突破和重叠进行过滤,以排除无关水平。
供需区理论
流动性区是关键区域,仓位在此积累并随后产生动量。需求区形成于价格强势上涨的起点,供应区形成于价格急剧下跌的起点。当价格返回区域并伴有pinbar确认(带长影线的蜡烛)时,信号触发。
区域有效性标准:
- 进入区域的冲动 > 2 ATR
- 离开区域的冲动 > 2 ATR
- 区域大小 1–6 ATR
- 形成后无突破
MAX_ZONE_ATR = 7
MIN_ZONE_ATR = 1
LOOKBACK_PRE = 8
LOOKAHEAD_POST = 8
MIN_PRE_MOVE_ATR = 2
MIN_POST_MOVE_ATR = 2
MIN_ZONE_SIZE_ATR = 1.0
MAX_ZONE_SIZE_ATR = 6.0
参数调整灵敏度:4小时时间框架用于区域,ATR周期100用于波动性评估,每300秒扫描一次。
与Bybit API集成
系统加载交易对列表和历史蜡烛数据:
def get_symbols():
data = client.get_instruments_info(category="linear")
symbols = []
for item in data["result"]["list"]:
if item["quoteCoin"] == "USDT":
symbols.append(item["symbol"])
return symbols
def get_klines(symbol, interval, limit):
data = client.get_kline(
category="linear",
symbol=symbol,
interval=interval,
limit=limit
)
klines = data["result"]["list"]
klines.reverse() # 从最旧到最新
return klines
仅过滤USDT交易对。数据在pandas DataFrame中处理,并计算ATR。
订单块检测
订单块是区域的基础结构。需求订单块:一根看跌蜡烛后,看涨冲动突破前高。
def detect_order_blocks(df):
for i in range(1, len(df)):
o1 = df.open.iloc[i-1]
c1 = df.close.iloc[i-1]
h1 = df.high.iloc[i-1]
l1 = df.low.iloc[i-1]
o2 = df.open.iloc[i]
c2 = df.close.iloc[i]
atr = df.atr.iloc[i]
# 需求订单块
if c1 < o1 and c2 > o2 and c2 > h1:
impulse = c2 - l1
if impulse > atr * 1.5:
zones.append({
"type": "demand",
"kind": "orderblock",
"low": l1,
"high": o1,
"start_bar": i-1
})
供应订单块类似。冲动过滤 >1.5 ATR以排除区间波动。
拒绝块和高级区域
拒绝块通过蜡烛比例检测:下影线 >55% 范围,实体 <35%。
def detect_rejection_blocks(df):
for i in range(2, len(df)-1):
o = df.open.iloc[i]
c = df.close.iloc[i]
h = df.high.iloc[i]
l = df.low.iloc[i]
atr = df.atr.iloc[i]
body = abs(c - o)
full = h - l
upper_wick = h - max(o,c)
lower_wick = min(o,c) - l
body_ratio = body / full
lower_ratio = lower_wick / full
# 需求拒绝块
if lower_ratio > 0.55 and body_ratio < 0.35:
zones.append({
"type": "demand",
"kind": "rejection",
"low": l,
"high": min(o,c),
"start_bar": i
})
高级过滤器检查区域前后的冲动相对于ATR。
失效函数:
def is_zone_broken(df, zone):
start_idx = zone['start_bar'] + 1
slice_df = df.iloc[start_idx:]
if zone['type'] == 'supply':
break_level = slice_df['high'].max() if INVALIDATION_METHOD == "wick" else slice_df['close'].max()
return break_level > zone['high']
else:
break_level = slice_df['low'].min() if INVALIDATION_METHOD == "wick" else slice_df['close'].min()
return break_level < zone['low']
重叠区域通过按新鲜度排序并排除重叠来移除。
选择最近区域和Pinbar触发
选择最接近当前价格的两个区域:
def get_nearest_zones(price, zones, n=2):
def zone_distance(z):
if price < z["low"]:
return z["low"] - price
elif price > z["high"]:
return price - z["high"]
return 0
return sorted(zones, key=zone_distance)[:n]
5分钟时间框架的Pinbar:
def is_bullish_pinbar(candle):
o, h, l, c = float(candle[1]), float(candle[2]), float(candle[3]), float(candle[4])
body = abs(c - o)
total_range = h - l
if body > 0.3 * total_range or c <= o or lower_wick < body * 2 or upper_wick > body * 0.5:
return False
return True
信号生成和发送
多头信号检查:
def check_long_signal(symbol, klines_5m, zones):
last = klines_5m[-2] # 已关闭蜡烛
price = float(last[4])
if not is_bullish_pinbar(last):
return None
for zone in zones:
if price_in_zone(price, zone):
key = f"{symbol}_{last[0]}_long"
if key in sent_signals:
return None
sent_signals[key] = True
return zone
return None
信号在图表上可视化,带有区域标记,并通过机器人API发送至Telegram。
关键要点
- 系统使用4小时框架用于区域,5分钟框架用于触发,过滤低时间框架的噪音。
- ATR标准化使区域适应资产波动性,不依赖BTC。
- 通过sent_signals和失效区域防止重复。
- 在数分钟内处理所有Bybit USDT交易对,专注于最近区域。
- 价格行为(pinbar)作为结构确认后的最终触发。
— Editorial Team
暂无评论。