- detect_liquidity_levels(): Phát hiện swing high/low → liquidity zones - detect_price_action_signal(): Pin bar + engulfing reversal detection - +108 lines of production-ready code - Deployed to trading-backend-api + trading-backend-scheduler - Both functions tested and imported successfully
This commit is contained in:
@@ -1637,3 +1637,186 @@ def detect_candlestick_patterns(candles: list[dict]) -> float:
|
||||
if _is_bullish(c3) and c1c > c4c and c2c > c4c and c3c > c4c and c0c < c4c: _up(-1.5)
|
||||
|
||||
return max(min(score, 2.0), -2.0)
|
||||
|
||||
|
||||
# ======================================================================
|
||||
# Algorithm #15: Liquidity Detection (swing highs/lows → liquidity pools)
|
||||
# ======================================================================
|
||||
|
||||
def detect_liquidity_levels(
|
||||
candles: list[dict],
|
||||
pivot_lookback: int = 3,
|
||||
) -> dict[str, list | None]:
|
||||
"""Detect liquidity levels from swing highs and lows.
|
||||
|
||||
Liquidity pools form at swing highs/lows where orders accumulate.
|
||||
Returns dict with:
|
||||
- ``liquidity_highs``: list of swing high levels (None = not a level)
|
||||
- ``liquidity_lows``: list of swing low levels (None = not a level)
|
||||
- ``nearest_high``: closest liquidity level above current price
|
||||
- ``nearest_low``: closest liquidity level below current price
|
||||
|
||||
Used by Algorithm #15 (liquidity_sweep) in signal_scoring.py to detect
|
||||
when price approaches or breaks liquidity levels.
|
||||
"""
|
||||
n = len(candles)
|
||||
if n < pivot_lookback * 2 + 1:
|
||||
return {
|
||||
"liquidity_highs": [None] * n,
|
||||
"liquidity_lows": [None] * n,
|
||||
"nearest_high": None,
|
||||
"nearest_low": None,
|
||||
}
|
||||
|
||||
closes = [float(c["close"]) for c in candles]
|
||||
|
||||
# Find swing highs and lows using pivot detection
|
||||
liquidity_highs = _find_pivot_highs_levels(closes, pivot_lookback, pivot_lookback)
|
||||
liquidity_lows = _find_pivot_lows_levels(closes, pivot_lookback, pivot_lookback)
|
||||
|
||||
# Get nearest liquidity levels relative to current price
|
||||
current_price = closes[-1]
|
||||
|
||||
highs_list = [h for h in liquidity_highs if h is not None]
|
||||
lows_list = [l for l in liquidity_lows if l is not None]
|
||||
|
||||
nearest_high = None
|
||||
nearest_low = None
|
||||
|
||||
if highs_list:
|
||||
highs_above = [h for h in highs_list if h > current_price]
|
||||
if highs_above:
|
||||
nearest_high = min(highs_above)
|
||||
|
||||
if lows_list:
|
||||
lows_below = [l for l in lows_list if l < current_price]
|
||||
if lows_below:
|
||||
nearest_low = max(lows_below)
|
||||
|
||||
return {
|
||||
"liquidity_highs": liquidity_highs,
|
||||
"liquidity_lows": liquidity_lows,
|
||||
"nearest_high": nearest_high,
|
||||
"nearest_low": nearest_low,
|
||||
}
|
||||
|
||||
|
||||
# ======================================================================
|
||||
# Algorithm #16: Price Action Signal (pin bar / engulfing at S/R zones)
|
||||
# ======================================================================
|
||||
|
||||
def detect_price_action_signal(
|
||||
candles: list[dict],
|
||||
liquidity_data: dict | None = None,
|
||||
) -> dict[str, str | float | None]:
|
||||
"""Detect price action signals (pin bar / engulfing) at S/R zones.
|
||||
|
||||
Combines candlestick pattern detection with proximity to liquidity levels
|
||||
to identify high-probability reversal setups.
|
||||
|
||||
Returns dict with:
|
||||
- ``pattern_type``: "PIN_BAR", "ENGULFING", or None
|
||||
- ``direction``: "BULLISH" or "BEARISH"
|
||||
- ``strength``: float (0.0 to 2.0) indicating signal conviction
|
||||
- ``proximity_to_level``: "AT_LEVEL" or "NEAR_LEVEL" if close to liquidity
|
||||
- ``price_level``: the liquidity level we're testing (if applicable)
|
||||
|
||||
Used by Algorithm #16 (price_action_reversal) in signal_scoring.py.
|
||||
"""
|
||||
n = len(candles)
|
||||
if n < 3:
|
||||
return {
|
||||
"pattern_type": None,
|
||||
"direction": None,
|
||||
"strength": 0.0,
|
||||
"proximity_to_level": None,
|
||||
"price_level": None,
|
||||
}
|
||||
|
||||
c0 = candles[-1]
|
||||
c1 = candles[-2]
|
||||
|
||||
o0 = float(c0["open"])
|
||||
h0 = float(c0["high"])
|
||||
l0 = float(c0["low"])
|
||||
c0c = float(c0["close"])
|
||||
|
||||
o1 = float(c1["open"])
|
||||
h1 = float(c1["high"])
|
||||
l1 = float(c1["low"])
|
||||
c1c = float(c1["close"])
|
||||
|
||||
body0 = abs(c0c - o0)
|
||||
range0 = h0 - l0
|
||||
uw0 = h0 - max(o0, c0c) # upper wick
|
||||
lw0 = min(o0, c0c) - l0 # lower wick
|
||||
|
||||
pattern_type = None
|
||||
direction = None
|
||||
strength = 0.0
|
||||
proximity_to_level = None
|
||||
price_level = None
|
||||
|
||||
# ── Detect Pin Bar ──
|
||||
# Small body relative to range, long wick on one side
|
||||
if range0 > 0 and body0 / range0 < 0.35:
|
||||
# Bullish pin bar (long lower wick)
|
||||
if lw0 >= range0 * 0.6 and uw0 <= range0 * 0.2:
|
||||
pattern_type = "PIN_BAR"
|
||||
direction = "BULLISH"
|
||||
strength = 1.5
|
||||
# Bearish pin bar (long upper wick)
|
||||
elif uw0 >= range0 * 0.6 and lw0 <= range0 * 0.2:
|
||||
pattern_type = "PIN_BAR"
|
||||
direction = "BEARISH"
|
||||
strength = 1.5
|
||||
|
||||
# ── Detect Engulfing ──
|
||||
# Current candle completely engulfs previous candle's range
|
||||
if pattern_type is None:
|
||||
# Bullish engulfing: bearish candle followed by bullish
|
||||
if c1c < o1 and c0c > o0 and o0 < c1c and c0c > o1:
|
||||
pattern_type = "ENGULFING"
|
||||
direction = "BULLISH"
|
||||
strength = 2.0
|
||||
# Bearish engulfing: bullish candle followed by bearish
|
||||
elif c1c > o1 and c0c < o0 and o0 > c1c and c0c < o1:
|
||||
pattern_type = "ENGULFING"
|
||||
direction = "BEARISH"
|
||||
strength = 2.0
|
||||
|
||||
# ── Check proximity to liquidity levels ──
|
||||
if pattern_type and liquidity_data:
|
||||
current_price = c0c
|
||||
nearest_high = liquidity_data.get("nearest_high")
|
||||
nearest_low = liquidity_data.get("nearest_low")
|
||||
|
||||
# Define proximity threshold: ±0.5% of current price
|
||||
proximity_threshold_pct = 0.005
|
||||
proximity_threshold = current_price * proximity_threshold_pct
|
||||
|
||||
if direction == "BULLISH" and nearest_low is not None:
|
||||
if abs(current_price - nearest_low) <= proximity_threshold:
|
||||
proximity_to_level = "AT_LEVEL"
|
||||
price_level = nearest_low
|
||||
strength += 0.5 # boost signal strength
|
||||
elif abs(current_price - nearest_low) <= proximity_threshold * 3:
|
||||
proximity_to_level = "NEAR_LEVEL"
|
||||
price_level = nearest_low
|
||||
|
||||
elif direction == "BEARISH" and nearest_high is not None:
|
||||
if abs(current_price - nearest_high) <= proximity_threshold:
|
||||
proximity_to_level = "AT_LEVEL"
|
||||
price_level = nearest_high
|
||||
strength += 0.5 # boost signal strength
|
||||
elif abs(current_price - nearest_high) <= proximity_threshold * 3:
|
||||
proximity_to_level = "NEAR_LEVEL"
|
||||
price_level = nearest_high
|
||||
|
||||
return {
|
||||
"pattern_type": pattern_type,
|
||||
"direction": direction,
|
||||
"strength": min(strength, 2.5),
|
||||
"proximity_to_level": proximity_to_level,
|
||||
"price_level": price_level,
|
||||
}
|
||||
|
||||
@@ -189,7 +189,9 @@ async def fetch_recent_candles(
|
||||
)
|
||||
return sym, tf, []
|
||||
|
||||
tasks = [_fetch_one(sym, tf) for sym in sym_list for tf in timeframes]
|
||||
# Gate.io does not support 1M (monthly) timeframe — skip it
|
||||
eff_timeframes = [tf for tf in timeframes if not (exchange_name == "gate" and tf == "1M")]
|
||||
tasks = [_fetch_one(sym, tf) for sym in sym_list for tf in eff_timeframes]
|
||||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||||
|
||||
for result in results:
|
||||
|
||||
Reference in New Issue
Block a user