mirror of
https://github.com/artemium428/tvsignals-to-tg.git
synced 2026-09-15 17:16:21 +00:00
cdbda8fea3
Per-strategy Heryon accounts, reversal captions with previous-trade path on charts, and Telegram replies chained by ticker. Co-authored-by: Cursor <cursoragent@cursor.com>
364 lines
12 KiB
Python
364 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from app.indicators.common import (
|
|
IndicatorSignal,
|
|
atr,
|
|
calc_size,
|
|
ohlc_frame,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FvgParams:
|
|
engulf_cover: float = 0.85
|
|
breakout_n: int = 10
|
|
max_bars: int = 300
|
|
use_lock_until_sl: bool = True
|
|
use_energy: bool = False
|
|
ss_fast_len: int = 20
|
|
ss_slow_len: int = 50
|
|
energy_k: float = 0.15
|
|
atr_len: int = 14
|
|
sl_atr_n: float = 1.0
|
|
sl_live_r: float = 2.0
|
|
tp1_rr: float = 2.0
|
|
tp2_rr: float = 0.0
|
|
tp3_rr: float = 0.0
|
|
risk_usd: float = 40.0
|
|
visual_timeframe: str = "360"
|
|
|
|
|
|
def _calc_levels(
|
|
entry: float,
|
|
ind_sl: float,
|
|
is_long: bool,
|
|
sl_live_r: float,
|
|
tp1_rr: float,
|
|
tp2_rr: float,
|
|
tp3_rr: float,
|
|
) -> tuple[float, float | None, float | None, float | None]:
|
|
r = abs(entry - ind_sl)
|
|
live_sl = entry - r * sl_live_r if is_long else entry + r * sl_live_r
|
|
|
|
def _tp(rr: float) -> float | None:
|
|
if rr <= 0:
|
|
return None
|
|
return entry + r * rr if is_long else entry - r * rr
|
|
|
|
return live_sl, _tp(tp1_rr), _tp(tp2_rr), _tp(tp3_rr)
|
|
|
|
|
|
def evaluate_fvg(df_ohlc: pd.DataFrame, params: FvgParams | None = None) -> pd.DataFrame:
|
|
"""FVG + engulf/breakout matching live Pine (lock until SL, same-dir supersede)."""
|
|
p = params or FvgParams()
|
|
df = ohlc_frame(df_ohlc)
|
|
n = len(df)
|
|
if n == 0:
|
|
return df
|
|
|
|
open_ = df["open"].to_numpy(dtype=float)
|
|
high = df["high"].to_numpy(dtype=float)
|
|
low = df["low"].to_numpy(dtype=float)
|
|
close = df["close"].to_numpy(dtype=float)
|
|
|
|
atr_s = atr(df["high"], df["low"], df["close"], p.atr_len)
|
|
atr_ok = np.nan_to_num(atr_s.to_numpy(dtype=float), nan=0.0)
|
|
|
|
long_energy_ok = np.ones(n, dtype=bool)
|
|
short_energy_ok = np.ones(n, dtype=bool)
|
|
|
|
prior_high_n = (
|
|
df["high"].shift(1).rolling(p.breakout_n, min_periods=p.breakout_n).max().to_numpy(dtype=float)
|
|
)
|
|
prior_low_n = (
|
|
df["low"].shift(1).rolling(p.breakout_n, min_periods=p.breakout_n).min().to_numpy(dtype=float)
|
|
)
|
|
|
|
prev_body = np.empty(n, dtype=float)
|
|
prev_body[0] = np.nan
|
|
prev_body[1:] = np.abs(close[:-1] - open_[:-1])
|
|
prev_bearish = np.zeros(n, dtype=bool)
|
|
prev_bullish = np.zeros(n, dtype=bool)
|
|
prev_bearish[1:] = close[:-1] < open_[:-1]
|
|
prev_bullish[1:] = close[:-1] > open_[:-1]
|
|
bull_min_cover = np.empty(n, dtype=float)
|
|
bear_min_cover = np.empty(n, dtype=float)
|
|
bull_min_cover[0] = np.nan
|
|
bear_min_cover[0] = np.nan
|
|
bull_min_cover[1:] = close[:-1] + prev_body[1:] * p.engulf_cover
|
|
bear_min_cover[1:] = close[:-1] - prev_body[1:] * p.engulf_cover
|
|
|
|
bull_engulf = (close > open_) & prev_bearish & (close >= bull_min_cover)
|
|
bear_engulf = (close < open_) & prev_bullish & (close <= bear_min_cover)
|
|
bull_bo = close > prior_high_n
|
|
bear_bo = close < prior_low_n
|
|
|
|
sl_long_px = close - atr_ok * p.sl_atr_n
|
|
sl_short_px = close + atr_ok * p.sl_atr_n
|
|
|
|
accepted_long = np.zeros(n, dtype=bool)
|
|
accepted_short = np.zeros(n, dtype=bool)
|
|
entry_px = np.full(n, np.nan)
|
|
long_sl_out = np.full(n, np.nan)
|
|
short_sl_out = np.full(n, np.nan)
|
|
long_tp1 = np.full(n, np.nan)
|
|
long_tp2 = np.full(n, np.nan)
|
|
long_tp3 = np.full(n, np.nan)
|
|
short_tp1 = np.full(n, np.nan)
|
|
short_tp2 = np.full(n, np.nan)
|
|
short_tp3 = np.full(n, np.nan)
|
|
|
|
bots: list[float] = []
|
|
tops: list[float] = []
|
|
dirs: list[int] = []
|
|
created: list[int] = []
|
|
mid_lo: list[float] = []
|
|
mid_hi: list[float] = []
|
|
valid: list[bool] = []
|
|
|
|
long_locked = False
|
|
short_locked = False
|
|
lock_long_sl = np.nan
|
|
lock_short_sl = np.nan
|
|
lock_long_bar = -1
|
|
lock_short_bar = -1
|
|
pos_dir = 0
|
|
pos_sl = np.nan
|
|
pos_bar = -1
|
|
|
|
last_long_sl = np.nan
|
|
last_long_tp1 = np.nan
|
|
last_long_tp2 = np.nan
|
|
last_long_tp3 = np.nan
|
|
last_short_sl = np.nan
|
|
last_short_tp1 = np.nan
|
|
last_short_tp2 = np.nan
|
|
last_short_tp3 = np.nan
|
|
last_entry = np.nan
|
|
|
|
def _remove(idx: int) -> None:
|
|
del bots[idx], tops[idx], dirs[idx], created[idx], mid_lo[idx], mid_hi[idx], valid[idx]
|
|
|
|
def _add(direction: int, bot: float, top: float, mlo: float, mhi: float, bar: int) -> None:
|
|
for k in range(len(dirs)):
|
|
if dirs[k] == direction:
|
|
valid[k] = False
|
|
bots.append(bot)
|
|
tops.append(top)
|
|
dirs.append(direction)
|
|
created.append(bar)
|
|
mid_lo.append(mlo)
|
|
mid_hi.append(mhi)
|
|
valid.append(True)
|
|
|
|
for i in range(n):
|
|
j = len(bots) - 1
|
|
while j >= 0:
|
|
if i - created[j] > p.max_bars:
|
|
_remove(j)
|
|
j -= 1
|
|
|
|
for k in range(len(bots)):
|
|
if not valid[k]:
|
|
continue
|
|
broken = close[i] < mid_lo[k] if dirs[k] == 1 else close[i] > mid_hi[k]
|
|
if broken:
|
|
valid[k] = False
|
|
|
|
new_bull = i >= 2 and low[i] > high[i - 2]
|
|
new_bear = i >= 2 and high[i] < low[i - 2]
|
|
if new_bull:
|
|
_add(1, float(high[i - 2]), float(low[i]), float(low[i - 1]), float(high[i - 1]), i)
|
|
if new_bear:
|
|
_add(-1, float(high[i]), float(low[i - 2]), float(low[i - 1]), float(high[i - 1]), i)
|
|
if (new_bull or new_bear) and bots:
|
|
last = len(bots) - 1
|
|
broken_new = close[i] < mid_lo[last] if dirs[last] == 1 else close[i] > mid_hi[last]
|
|
if broken_new:
|
|
valid[last] = False
|
|
|
|
sl_long_hit = pos_dir == 1 and i > pos_bar and low[i] <= pos_sl
|
|
sl_short_hit = pos_dir == -1 and i > pos_bar and high[i] >= pos_sl
|
|
if sl_long_hit or sl_short_hit:
|
|
pos_dir = 0
|
|
pos_sl = np.nan
|
|
pos_bar = -1
|
|
if p.use_lock_until_sl:
|
|
if sl_long_hit or (long_locked and i > lock_long_bar and low[i] <= lock_long_sl):
|
|
long_locked = False
|
|
lock_long_sl = np.nan
|
|
lock_long_bar = -1
|
|
if sl_short_hit or (short_locked and i > lock_short_bar and high[i] >= lock_short_sl):
|
|
short_locked = False
|
|
lock_short_sl = np.nan
|
|
lock_short_bar = -1
|
|
|
|
cur_idx = -1
|
|
for k in range(len(bots) - 1, -1, -1):
|
|
if valid[k]:
|
|
cur_idx = k
|
|
break
|
|
|
|
sig_long = False
|
|
sig_short = False
|
|
if cur_idx >= 0:
|
|
cur_dir = dirs[cur_idx]
|
|
cur_bot = bots[cur_idx]
|
|
cur_top = tops[cur_idx]
|
|
opp_ok = True
|
|
prev = cur_idx - 1
|
|
while prev >= 0:
|
|
if dirs[prev] != cur_dir:
|
|
opp_ok = not valid[prev]
|
|
break
|
|
prev -= 1
|
|
|
|
if opp_ok:
|
|
long_free = (not p.use_lock_until_sl) or (not long_locked)
|
|
short_free = (not p.use_lock_until_sl) or (not short_locked)
|
|
long_ok = cur_dir == 1 and close[i] > cur_bot
|
|
short_ok = cur_dir == -1 and close[i] < cur_top
|
|
long_pat = long_ok and long_free and (bool(bull_engulf[i]) or bool(bull_bo[i]))
|
|
short_pat = short_ok and short_free and (bool(bear_engulf[i]) or bool(bear_bo[i]))
|
|
sig_long = long_pat and bool(long_energy_ok[i])
|
|
sig_short = short_pat and bool(short_energy_ok[i])
|
|
|
|
if sig_long:
|
|
entry = float(close[i])
|
|
ind_sl = float(sl_long_px[i])
|
|
if np.isnan(ind_sl) or ind_sl >= entry:
|
|
sig_long = False
|
|
else:
|
|
live_sl, t1, t2, t3 = _calc_levels(
|
|
entry, ind_sl, True, p.sl_live_r, p.tp1_rr, p.tp2_rr, p.tp3_rr
|
|
)
|
|
last_long_sl, last_long_tp1, last_long_tp2, last_long_tp3 = (
|
|
live_sl, t1, t2 if t2 is not None else np.nan, t3 if t3 is not None else np.nan
|
|
)
|
|
last_short_sl = last_short_tp1 = last_short_tp2 = last_short_tp3 = np.nan
|
|
last_entry = entry
|
|
pos_dir = 1
|
|
pos_sl = live_sl
|
|
pos_bar = i
|
|
if p.use_lock_until_sl:
|
|
long_locked = True
|
|
lock_long_sl = live_sl
|
|
lock_long_bar = i
|
|
short_locked = False
|
|
lock_short_sl = np.nan
|
|
lock_short_bar = -1
|
|
|
|
if sig_short:
|
|
entry = float(close[i])
|
|
ind_sl = float(sl_short_px[i])
|
|
if np.isnan(ind_sl) or ind_sl <= entry:
|
|
sig_short = False
|
|
else:
|
|
live_sl, t1, t2, t3 = _calc_levels(
|
|
entry, ind_sl, False, p.sl_live_r, p.tp1_rr, p.tp2_rr, p.tp3_rr
|
|
)
|
|
last_short_sl, last_short_tp1, last_short_tp2, last_short_tp3 = (
|
|
live_sl, t1, t2 if t2 is not None else np.nan, t3 if t3 is not None else np.nan
|
|
)
|
|
last_long_sl = last_long_tp1 = last_long_tp2 = last_long_tp3 = np.nan
|
|
last_entry = entry
|
|
pos_dir = -1
|
|
pos_sl = live_sl
|
|
pos_bar = i
|
|
if p.use_lock_until_sl:
|
|
short_locked = True
|
|
lock_short_sl = live_sl
|
|
lock_short_bar = i
|
|
long_locked = False
|
|
lock_long_sl = np.nan
|
|
lock_long_bar = -1
|
|
|
|
accepted_long[i] = sig_long
|
|
accepted_short[i] = sig_short
|
|
entry_px[i] = last_entry
|
|
long_sl_out[i] = last_long_sl
|
|
short_sl_out[i] = last_short_sl
|
|
long_tp1[i] = last_long_tp1
|
|
long_tp2[i] = last_long_tp2
|
|
long_tp3[i] = last_long_tp3
|
|
short_tp1[i] = last_short_tp1
|
|
short_tp2[i] = last_short_tp2
|
|
short_tp3[i] = last_short_tp3
|
|
|
|
df["atr"] = atr_s
|
|
df["buy_sig"] = accepted_long
|
|
df["sell_sig"] = accepted_short
|
|
df["entry_px"] = entry_px
|
|
df["long_sl"] = long_sl_out
|
|
df["short_sl"] = short_sl_out
|
|
df["long_tp1"] = long_tp1
|
|
df["long_tp2"] = long_tp2
|
|
df["long_tp3"] = long_tp3
|
|
df["short_tp1"] = short_tp1
|
|
df["short_tp2"] = short_tp2
|
|
df["short_tp3"] = short_tp3
|
|
return df
|
|
|
|
|
|
def last_fvg_signal(df: pd.DataFrame, params: FvgParams | None = None) -> IndicatorSignal | None:
|
|
if df.empty:
|
|
return None
|
|
p = params or FvgParams()
|
|
last = df.iloc[-1]
|
|
ts = int(df.index[-1].timestamp())
|
|
close = float(last["close"])
|
|
|
|
def _opt(value: object) -> float | None:
|
|
if value is None or (isinstance(value, float) and np.isnan(value)):
|
|
return None
|
|
number = float(value)
|
|
return None if np.isnan(number) else number
|
|
|
|
if bool(last["buy_sig"]):
|
|
sl = float(last["long_sl"])
|
|
tp1 = float(last["long_tp1"])
|
|
if np.isnan(sl) or np.isnan(tp1) or np.isnan(close):
|
|
return None
|
|
size = calc_size(close, sl, p.risk_usd)
|
|
if np.isnan(size):
|
|
return None
|
|
return IndicatorSignal(
|
|
strategy_id="fvg",
|
|
side="long",
|
|
entry=close,
|
|
close=close,
|
|
sl=sl,
|
|
tp1=tp1,
|
|
tp2=_opt(last["long_tp2"]),
|
|
tp3=_opt(last["long_tp3"]),
|
|
size_usd=size,
|
|
bar_open_ts=ts,
|
|
visual_timeframe=p.visual_timeframe,
|
|
)
|
|
if bool(last["sell_sig"]):
|
|
sl = float(last["short_sl"])
|
|
tp1 = float(last["short_tp1"])
|
|
if np.isnan(sl) or np.isnan(tp1) or np.isnan(close):
|
|
return None
|
|
size = calc_size(close, sl, p.risk_usd)
|
|
if np.isnan(size):
|
|
return None
|
|
return IndicatorSignal(
|
|
strategy_id="fvg",
|
|
side="short",
|
|
entry=close,
|
|
close=close,
|
|
sl=sl,
|
|
tp1=tp1,
|
|
tp2=_opt(last["short_tp2"]),
|
|
tp3=_opt(last["short_tp3"]),
|
|
size_usd=size,
|
|
bar_open_ts=ts,
|
|
visual_timeframe=p.visual_timeframe,
|
|
)
|
|
return None
|