From cdbda8fea3a164009432f0d594a9d92278fa6f06 Mon Sep 17 00:00:00 2001 From: Artemii Peretiachenko Date: Sun, 30 Aug 2026 20:37:16 +0200 Subject: [PATCH] Replace TradingView polling with a local LTF/FVG scanner that posts Telegram cards and Heryon webhooks. Per-strategy Heryon accounts, reversal captions with previous-trade path on charts, and Telegram replies chained by ticker. Co-authored-by: Cursor --- .env.example | 19 ++ .gitignore | 2 + Dockerfile | 3 + README.md | 159 +++++++-------- app/binance.py | 117 ++++++++++- app/chart.py | 156 +++++++++++++-- app/config.py | 25 +++ app/formatter.py | 47 +++-- app/heryon.py | 81 ++++++++ app/indicators/__init__.py | 11 ++ app/indicators/common.py | 236 +++++++++++++++++++++++ app/indicators/fvg.py | 364 +++++++++++++++++++++++++++++++++++ app/indicators/ltf.py | 384 +++++++++++++++++++++++++++++++++++++ app/main.py | 103 +++------- app/models.py | 25 ++- app/pipeline.py | 181 +++++++++++++++++ app/scanner.py | 164 ++++++++++++++++ app/state.py | 145 ++++++++++++++ app/telegram.py | 21 ++ app/watchlist.py | 54 ++++++ data/.gitkeep | 0 docker-compose.yml | 3 + requirements.txt | 2 + watchlist.yaml | 19 ++ 24 files changed, 2125 insertions(+), 196 deletions(-) create mode 100644 app/heryon.py create mode 100644 app/indicators/__init__.py create mode 100644 app/indicators/common.py create mode 100644 app/indicators/fvg.py create mode 100644 app/indicators/ltf.py create mode 100644 app/pipeline.py create mode 100644 app/scanner.py create mode 100644 app/state.py create mode 100644 app/watchlist.py create mode 100644 data/.gitkeep create mode 100644 watchlist.yaml diff --git a/.env.example b/.env.example index 62e6322..4582e85 100644 --- a/.env.example +++ b/.env.example @@ -1,7 +1,26 @@ TELEGRAM_BOT_TOKEN=123456:ABC-DEF TELEGRAM_CHAT_ID=-1001234567890 # Default forum topic when webhook URL has no /{thread_id} segment +# and watchlist strategy telegram_thread_id is null TELEGRAM_MESSAGE_THREAD_ID=1 WEBHOOK_SECRET=change-me-to-a-long-random-string HOST=0.0.0.0 PORT=8000 + +# Local scanner (replaces TradingView alerts) +SCANNER_ENABLED=true +SCANNER_POLL_SECONDS=20 +WATCHLIST_PATH=watchlist.yaml +SCANNER_STATE_PATH=data/scanner.db +RISK_USD=40 + +# Heryon / Geryon — same JSON as Pine alertcondition (leave URL empty to skip) +GERYON_WEBHOOK_URL= +GERYON_ORDER_TYPE=bbo +# Per-strategy account/secret (fallback: GERYON_ACCOUNT_ID / GERYON_SECRET) +GERYON_ACCOUNT_ID_LTF= +GERYON_SECRET_LTF= +GERYON_ACCOUNT_ID_FVG= +GERYON_SECRET_FVG= +GERYON_ACCOUNT_ID=admin1 +GERYON_SECRET= diff --git a/.gitignore b/.gitignore index 370dec5..36eb93b 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,5 @@ __pycache__/ .pytest_cache/ .mypy_cache/ .DS_Store +data/*.db +data/*.db-* diff --git a/Dockerfile b/Dockerfile index c90377b..5523631 100644 --- a/Dockerfile +++ b/Dockerfile @@ -12,6 +12,9 @@ COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY app ./app +COPY watchlist.yaml . + +RUN mkdir -p /app/data ENV PYTHONUNBUFFERED=1 ENV MPLBACKEND=Agg diff --git a/README.md b/README.md index abc0f9f..971dc4a 100644 --- a/README.md +++ b/README.md @@ -1,10 +1,12 @@ -# TradingView → Telegram setup service +# Signals → Telegram + Heryon -Accepts TradingView webhook alerts, renders a Binance Futures candlestick setup chart, and posts photo + caption into a Telegram forum topic. +Runs the 15m LTF and 6h FVG indicators on a Binance Futures watchlist, posts setup charts to a Telegram forum topic, and forwards the same Heryon webhook JSON that TradingView `alertcondition` used to send. + +Inbound `POST /h/...` is still accepted for manual/debug payloads. Those requests are **not** forwarded to Heryon (avoids doubles if TradingView is still on). ## Quick start (Docker / VPS) -1. Copy env and fill Telegram values: +1. Copy env and fill Telegram + Heryon values: ```bash cp .env.example .env @@ -14,33 +16,33 @@ cp .env.example .env TELEGRAM_BOT_TOKEN=... TELEGRAM_CHAT_ID=-100... TELEGRAM_MESSAGE_THREAD_ID=... -HOST=0.0.0.0 -PORT=8000 +WEBHOOK_SECRET=... +GERYON_WEBHOOK_URL=http://geryon:PORT/path # empty = Telegram only +GERYON_ORDER_TYPE=bbo +GERYON_ACCOUNT_ID_LTF=... +GERYON_SECRET_LTF=... +GERYON_ACCOUNT_ID_FVG=... +GERYON_SECRET_FVG=... ``` -2. Build and run: +2. Edit [`watchlist.yaml`](watchlist.yaml): symbols and optional `telegram_thread_id` per strategy (15m vs 6h topics). + +3. Build and run: ```bash +mkdir -p data docker compose up -d --build ``` -3. Health check: +4. Health check: ```bash curl http://127.0.0.1:8000/health ``` -4. Put HTTPS in front (nginx/Caddy) and point TradingView webhook to: +On first start the scanner **primes** the last closed bar per symbol and does not replay history. The next closed 15m / 6h bar can fire a signal. -`https://your-domain/h/` - -or, for a specific forum topic: - -`https://your-domain/h//` - -Same alert message body; different URLs → different topics (e.g. one alert per timeframe). Without a thread segment, the env `TELEGRAM_MESSAGE_THREAD_ID` is used. - -Bot must be added to the group/forum and allowed to post in the target topic. +After it is stable, turn off the TradingView alerts that used to hit this service and Heryon. ## Local run (without Docker) @@ -49,18 +51,53 @@ python -m venv .venv source .venv/bin/activate pip install -r requirements.txt cp .env.example .env # fill values +mkdir -p data uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload ``` -## TradingView alert JSON +## Scanner -The webhook accepts a **single-line or pretty-printed JSON** body, including TradingView’s common `Content-Type: text/plain`. +| Strategy | Timeframe | Risk | Heryon TPs | +|---|---|---|---| +| `ltf` (`428th_v16_LTF`) | 15m + 6h filters | SL = 16h low/high; TP 1.75 / 2 / 4 R | `tp1/2/3_fix` 30% / 30% / 40%, `sl_to_bk=tp1` | +| `fvg` (`428th_v16`) | 6h | live SL 2R from ATR×1 stick; TP1 2R | `tp1_fix` 30%, empty tp2/tp3 (leftover until reverse) | -Static example (alert **Webhook message** body) — primary setup (`signal_sequence: 1`): +- Poll interval: `SCANNER_POLL_SECONDS` (default 20). Only **closed** candles are evaluated. +- State: `data/scanner.db` (last bar + Heryon nonces). Restart does not re-send the primed bar. +- Same-side lock matches live Pine (`open_side` on LTF, `use_lock_until_sl` on FVG). +- Tickers in Telegram / Heryon: `{SYMBOL}.P` (e.g. `BTCUSDT.P`). +- Size: `round(RISK_USD / abs(entry−sl)/entry)` with `RISK_USD=40`. + +Set `SCANNER_ENABLED=false` to run as a webhook-only Telegram bridge. + +### Heryon payload (LTF) ```json { - "ticker": "{{ticker}}", + "action": "buy", + "ticker": "BTCUSDT.P", + "account_id": "admin1", + "order_type": "bbo", + "position_size_usd": "5000", + "stop_loss": "...", + "tp1": "...", "tp1_fix": "30%", + "tp2": "...", "tp2_fix": "30%", + "tp3": "...", "tp3_fix": "40%", + "sl_to_bk": "tp1", + "nonce": "BTCUSDT.P-limit-long-1721736000", + "secret": "..." +} +``` + +FVG omits `tp2` / `tp3` and their `_fix` fields. Empty string keys are not sent. + +## Manual / debug webhook + +`POST /h/` or `POST /h//` still accepts the Telegram JSON body (including TradingView `text/plain`). Telegram only — no Heryon forward. + +```json +{ + "ticker": "BTCUSDT.P", "action": "long", "entry_price": "65034.7", "current_price": "65034.7", @@ -74,84 +111,36 @@ Static example (alert **Webhook message** body) — primary setup (`signal_seque } ``` -Control update (`signal_sequence` > 1) — Pine freezes `entry_price` / SL / TPs / `signal_time` from seq 1 and sends live `current_price`: +`take_profit_2_price` / `take_profit_3_price` are optional (omit or `""` for TP1-only FVG-style setups). -```json -{ - "ticker": "{{ticker}}", - "action": "short", - "entry_price": "65034.7", - "current_price": "64000.1", - "stop_loss_price": "65896.8", - "take_profit_1_price": "63904.9", - "take_profit_2_price": "62781.6", - "take_profit_3_price": "61635.3", - "visual_timeframe": "15", - "signal_sequence": 2, - "signal_time": 1721736000 -} -``` - -### PineScript `alert()` (recommended) - -Build the JSON inside `alert()`. A continuous one-line string is fine. - -In the TradingView alert dialog: - -- Webhook URL: `https://your-domain/h/` or `https://your-domain/h//` (per-topic; same message body) -- Message: only `{{alert_message}}` (do not paste a second JSON next to it) - -On **seq == 1**: store `entry_price = close`, freeze SL/TPs, and `signal_time = time / 1000` (bar open, unix seconds). On **seq > 1**: keep those frozen fields; only refresh `current_price` (= live `close`). Example shape: - -```pinescript -alert('{"ticker":"' + syminfo.ticker + '","action":"long","entry_price":"' + str.tostring(long_entry_price, format.mintick) + '","current_price":"' + str.tostring(close, format.mintick) + '","stop_loss_price":"' + str.tostring(long_sl_price, format.mintick) + '","take_profit_1_price":"' + str.tostring(long_tp1_price, format.mintick) + '","take_profit_2_price":"' + str.tostring(long_tp2_price, format.mintick) + '","take_profit_3_price":"' + str.tostring(long_tp3_price, format.mintick) + '","visual_timeframe":"' + timeframe.period + '","signal_sequence":' + str.tostring(buyCount) + ',"signal_time":' + str.tostring(signal_buy_time) + '}', alert.freq_once_per_bar_close) -``` - -(Same idea for shorts with `short_entry_price` / `sellCount` / `signal_sell_time`.) Caption emoji/labels are built by the service from `action` + `signal_sequence` — do not put them in the webhook JSON. - -Field notes: +Caption emoji/labels are built from `action` + `signal_sequence`. | Field | Description | |---|---| -| `ticker` | any common TV form (`BTCUSDT.P`, `BTCUSDT`, `BINANCE:ETHUSDT`, `BTC/USDT`, …) → normalized to Binance Futures symbol for the chart; caption keeps the original | +| `ticker` | TV form (`BTCUSDT.P`, `BINANCE:ETHUSDT`, …) → Binance Futures for the chart; caption keeps the original | | `action` | `long` or `short` | -| `entry_price` | trade entry from seq 1 (equals `current_price` on primary signal) | -| `current_price` | live price (`close` at alert time) | -| `*_price` | strings with your display precision | -| `visual_timeframe` | `1`, `3`, `5`, `15`, `30`, `60`, `120`, `240`, `D`, `W` (also `15m`, `1h`, …) | -| `signal_sequence` | `1` = primary setup; `>1` = control update of that trade | -| `signal_time` | unix seconds of the seq-1 bar open (UTC); chart draws Entry/SL/TP zones from that candle | +| `*_price` | strings; TP2/TP3 optional | +| `visual_timeframe` | `15`, `360`, `15m`, `6h`, … | +| `signal_sequence` | `1` = setup; `>1` = control update | +| `signal_time` | unix seconds of the seq-1 bar open (UTC) | ## Caption format -**seq `1` (setup):** +**seq `1` (setup):** `💚 Buy` / `💔 Sell` — Price / SL (risk %) / TP1 and TP2/TP3 when present. -- long: `BTCUSDT.P 💚 Buy` -- short: `BTCUSDT.P 💔 Sell` -- body: `Price` / `SL (risk %)` / `TP1–3` +**seq `>1` (control):** `🌱 Buy Seq: N` / `🥀 Sell Seq: N` — Entry / live Price / `Current profit`. -**seq `>1` (control):** - -- long: `BTCUSDT.P 🌱 Buy Seq: N` -- short: `BTCUSDT.P 🥀 Sell Seq: N` -- body: `Entry price` (from seq 1) / live `Price` / `Current profit: +1.6% (RR 1:1.2)` -- no new SL/TP lines in the caption - -Prices are shown with `$` and thousand spaces (`65034.7` → `$65 034.7`). -For seq 1, SL includes distance from entry: `SL: $63 904.9 (-2.37%)` (risk %, negative for both long and short). -For seq >1, profit % is signed vs entry; RR is `|price−entry| / |entry−SL|` with the same sign as profit. +Prices: `$` and thousand spaces (`65034.7` → `$65 034.7`). ## Behavior -1. Validate secret + payload, then immediately respond `200` (`{"ok": true, "accepted": true}`) so TradingView does not time out -2. In the background: fetch ~90 klines from Binance USDT-M Futures (public, no API key) -3. Render PNG: candles + Entry / SL / TP1–3 from the payload, starting at the `signal_time` candle (seq `>1` reuses frozen seq-1 levels/time and also marks live `Price`) -4. `sendPhoto` to `TELEGRAM_CHAT_ID` topic from URL path or env `TELEGRAM_MESSAGE_THREAD_ID` -5. If chart/klines fail → text-only `sendMessage` fallback -6. If Telegram fails → logged only (HTTP already returned `200`) +1. Scanner (if enabled): closed bar → indicator → Telegram chart + Heryon JSON +2. Inbound webhook: validate secret, `200` immediately, background Telegram only +3. Chart: ~90 USDT-M klines; on failure, text-only Telegram +4. Telegram or Heryon failure is logged (webhook HTTP already returned `200`) ## Endpoints - `GET /health` → `{"status":"ok"}` -- `POST /h/` → signal payload above; topic from env `TELEGRAM_MESSAGE_THREAD_ID` (wrong/missing secret → `404`) -- `POST /h//` → same payload; topic from path (`thread_id` must be `>= 1`) +- `POST /h/` → debug payload; topic from `TELEGRAM_MESSAGE_THREAD_ID` +- `POST /h//` → same; topic from path (`thread_id` >= 1) diff --git a/app/binance.py b/app/binance.py index df5771d..5f76c96 100644 --- a/app/binance.py +++ b/app/binance.py @@ -8,7 +8,25 @@ import pandas as pd logger = logging.getLogger(__name__) BINANCE_FUTURES_KLINES_URL = "https://fapi.binance.com/fapi/v1/klines" +BINANCE_EXCHANGE_INFO_URL = "https://fapi.binance.com/fapi/v1/exchangeInfo" DEFAULT_LIMIT = 90 +# 6h charts use 1/3 of the default window so price action looks closer. +CHART_LIMIT_BY_INTERVAL: dict[str, int] = { + "15m": 135, + "6h": 75, +} + + +def chart_kline_limit(interval: str) -> int: + return CHART_LIMIT_BY_INTERVAL.get(interval, DEFAULT_LIMIT) + + +def chart_right_pad(interval: str) -> int: + """Empty candles to the right; scale with the visible window.""" + if interval == "6h": + return 12 + return 15 + # TradingView-style timeframe → Binance Futures interval TIMEFRAME_MAP: dict[str, str] = { @@ -95,6 +113,45 @@ def to_binance_interval(visual_timeframe: str) -> str: return interval +_INTERVAL_DELTA: dict[str, pd.Timedelta] = { + "1m": pd.Timedelta(minutes=1), + "3m": pd.Timedelta(minutes=3), + "5m": pd.Timedelta(minutes=5), + "15m": pd.Timedelta(minutes=15), + "30m": pd.Timedelta(minutes=30), + "1h": pd.Timedelta(hours=1), + "2h": pd.Timedelta(hours=2), + "4h": pd.Timedelta(hours=4), + "6h": pd.Timedelta(hours=6), + "8h": pd.Timedelta(hours=8), + "12h": pd.Timedelta(hours=12), + "1d": pd.Timedelta(days=1), + "1w": pd.Timedelta(weeks=1), +} + + +def interval_timedelta(interval: str) -> pd.Timedelta: + key = interval if interval in _INTERVAL_DELTA else to_binance_interval(interval) + delta = _INTERVAL_DELTA.get(key) + if delta is None: + raise ValueError(f"Unsupported interval: {interval!r}") + return delta + + +def drop_forming_candles(df: pd.DataFrame, interval: str) -> pd.DataFrame: + """Drop the in-progress candle (open + interval > now).""" + if df.empty: + return df + delta = interval_timedelta(interval) + now = pd.Timestamp.now(tz="UTC") + idx = df.index + if idx.tz is None: + idx = idx.tz_localize("UTC") + else: + idx = idx.tz_convert("UTC") + return df.loc[idx + delta <= now] + + async def fetch_klines( symbol: str, interval: str, @@ -102,12 +159,16 @@ async def fetch_klines( limit: int = DEFAULT_LIMIT, end_ms: int | None = None, timeout: float = 15.0, + closed_only: bool = False, + client: httpx.AsyncClient | None = None, ) -> pd.DataFrame: """Fetch OHLCV klines from Binance USDT-M Futures. If ``end_ms`` is set, returns candles ending at/before that UTC epoch millis (useful for historical / as-of charts). """ + if limit > 1500: + raise ValueError("Binance klines limit is 1500") params: dict[str, str | int] = { "symbol": symbol, "interval": interval, @@ -115,10 +176,16 @@ async def fetch_klines( } if end_ms is not None: params["endTime"] = end_ms - async with httpx.AsyncClient(timeout=timeout) as client: - response = await client.get(BINANCE_FUTURES_KLINES_URL, params=params) + + http = client or httpx.AsyncClient(timeout=timeout) + own_client = client is None + try: + response = await http.get(BINANCE_FUTURES_KLINES_URL, params=params) response.raise_for_status() raw = response.json() + finally: + if own_client: + await http.aclose() if not raw: raise ValueError(f"Empty klines for {symbol} {interval}") @@ -146,6 +213,52 @@ async def fetch_klines( df = df.set_index("Date")[["open", "high", "low", "close", "volume"]] df.columns = ["Open", "High", "Low", "Close", "Volume"] df = df.dropna() + if closed_only: + df = drop_forming_candles(df, interval) + return df if df.empty: raise ValueError(f"No valid OHLCV rows for {symbol} {interval}") return df + + +_TICK_SIZE: dict[str, float] = {} + + +async def load_tick_sizes( + client: httpx.AsyncClient | None = None, + *, + timeout: float = 20.0, +) -> None: + """Cache Binance USDT-M PRICE_FILTER.tickSize per symbol (Pine mintick).""" + http = client or httpx.AsyncClient(timeout=timeout) + own = client is None + try: + response = await http.get(BINANCE_EXCHANGE_INFO_URL) + response.raise_for_status() + payload = response.json() + finally: + if own: + await http.aclose() + + ticks: dict[str, float] = {} + for item in payload.get("symbols") or []: + name = str(item.get("symbol") or "") + if not name: + continue + for filt in item.get("filters") or []: + if filt.get("filterType") == "PRICE_FILTER": + raw = filt.get("tickSize") + if raw is None: + continue + tick = float(raw) + if tick > 0: + ticks[name] = tick + break + if ticks: + _TICK_SIZE.update(ticks) + logger.info("Loaded tick sizes for %s symbols", len(ticks)) + + +def get_tick_size(symbol: str) -> float: + key = to_binance_symbol(symbol) + return _TICK_SIZE.get(key, 0.01) diff --git a/app/chart.py b/app/chart.py index 9650685..2a6fb3b 100644 --- a/app/chart.py +++ b/app/chart.py @@ -125,6 +125,91 @@ def _position_start_x(df: pd.DataFrame, signal_time: int | None) -> int: return min(pos, last_x) +def _draw_previous_trade( + ax, + df: pd.DataFrame, + *, + prev_entry: float, + prev_side: str, + prev_signal_time: int, + exit_price: float, + exit_x: int, +) -> None: + """Entry → exit of the trade that this reversal closes.""" + prev_x = _position_start_x(df, prev_signal_time) + if prev_x >= exit_x: + return + profitable = ( + exit_price >= prev_entry if prev_side == "long" else exit_price <= prev_entry + ) + color = COLORS["up"] if profitable else COLORS["down"] + y_low = min(prev_entry, exit_price) + y_high = max(prev_entry, exit_price) + height = y_high - y_low + if height <= 0: + height = abs(prev_entry) * 1e-6 or 1e-8 + ax.add_patch( + Rectangle( + (prev_x, y_low), + exit_x - prev_x, + height, + facecolor=color, + edgecolor="none", + alpha=0.04, + zorder=1, + ) + ) + closes = df["Close"].iloc[prev_x : exit_x + 1].astype(float) + xs = list(range(prev_x, prev_x + len(closes))) + ax.plot( + xs, + closes.to_numpy(), + color=color, + linewidth=0.9, + linestyle=(0, (3, 4)), + alpha=0.28, + zorder=5, + solid_capstyle="round", + ) + ax.plot( + [prev_x, exit_x], + [prev_entry, exit_price], + color=color, + linewidth=1.05, + linestyle=(0, (4, 5)), + alpha=0.45, + zorder=6, + ) + ax.scatter( + [prev_x], + [prev_entry], + s=22, + c=color, + marker="o", + zorder=7, + edgecolors="#ffffff", + linewidths=0.7, + ) + ax.annotate( + "Prev", + xy=(prev_x, prev_entry), + xytext=(6, 8), + textcoords="offset points", + va="bottom", + ha="left", + fontsize=7, + color=color, + zorder=8, + bbox={ + "boxstyle": "round,pad=0.2", + "facecolor": COLORS["label_bg"], + "edgecolor": color, + "linewidth": 0.6, + "alpha": 0.88, + }, + ) + + def render_setup_chart( df: pd.DataFrame, *, @@ -133,18 +218,23 @@ def render_setup_chart( entry: str, stop_loss: str, tp1: str, - tp2: str, - tp3: str, timeframe: str, + tp2: str | None = None, + tp3: str | None = None, current_price: str | None = None, signal_time: int | None = None, + right_pad: int | None = None, + prev_entry: str | None = None, + prev_side: str | None = None, + prev_signal_time: int | None = None, ) -> bytes: entry_p = _parse_price(entry) sl_p = _parse_price(stop_loss) tp1_p = _parse_price(tp1) - tp2_p = _parse_price(tp2) - tp3_p = _parse_price(tp3) + tp2_p = _parse_price(tp2) if tp2 else None + tp3_p = _parse_price(tp3) if tp3 else None current_p = _parse_price(current_price) if current_price is not None else None + prev_entry_p = _parse_price(prev_entry) if prev_entry else None is_long = action == "long" reward_rgb = COLORS["reward_fill"] @@ -154,13 +244,20 @@ def render_setup_chart( # Position tool starts at seq==1 candle; "now" is the last real candle now_x = len(df) - 1 entry_x = _position_start_x(df, signal_time) - plot_df = _pad_right(df, RIGHT_PAD_CANDLES) + plot_df = _pad_right(df, RIGHT_PAD_CANDLES if right_pad is None else right_pad) - level_prices = [entry_p, sl_p, tp1_p, tp2_p, tp3_p] + level_prices = [entry_p, sl_p, tp1_p] + if tp2_p is not None: + level_prices.append(tp2_p) + if tp3_p is not None: + level_prices.append(tp3_p) if current_p is not None: level_prices.append(current_p) + if prev_entry_p is not None: + level_prices.append(prev_entry_p) y_min = min(float(df["Low"].min()), *level_prices) y_max = max(float(df["High"].max()), *level_prices) + hold_remainder = tp2_p is None and tp3_p is None price_pad = (y_max - y_min) * 0.06 or y_max * 0.002 vol_scaled, vol_colors = _volume_overlay(plot_df, y_low=y_min, y_high=y_max) @@ -227,18 +324,22 @@ def render_setup_chart( # Room for volume bars under candles vol_floor = float(vol_scaled.dropna().min()) if vol_scaled.notna().any() else y_min ax.set_ylim(min(y_min - price_pad, vol_floor) - price_pad * 0.3, y_max + price_pad) + ylim_lo, ylim_hi = ax.get_ylim() + remainder_end = (ylim_hi if is_long else ylim_lo) if hold_remainder else None x_right = ax.get_xlim()[1] zone_width = x_right - entry_x # Levels + zones start at the entry (last real) candle, not full chart width - level_specs = [ + level_specs: list[tuple[float, str, str, float]] = [ (entry_p, COLORS["entry"], "--", 1.2), (sl_p, COLORS["sl"], "-", 1.2), (tp1_p, COLORS["tp1"], ":", 1.0), - (tp2_p, COLORS["tp2"], ":", 1.0), - (tp3_p, COLORS["tp3"], ":", 1.0), ] + if tp2_p is not None: + level_specs.append((tp2_p, COLORS["tp2"], ":", 1.0)) + if tp3_p is not None: + level_specs.append((tp3_p, COLORS["tp3"], ":", 1.0)) for price, color, ls, lw in level_specs: ax.hlines( price, @@ -264,12 +365,17 @@ def render_setup_chart( ) ) - # Three reward bands with decreasing opacity toward farther TPs - reward_bands = ( + reward_bands: list[tuple[float, float, float]] = [ (entry_p, tp1_p, REWARD_ALPHAS[0]), - (tp1_p, tp2_p, REWARD_ALPHAS[1]), - (tp2_p, tp3_p, REWARD_ALPHAS[2]), - ) + ] + if tp2_p is not None: + reward_bands.append((tp1_p, tp2_p, REWARD_ALPHAS[1])) + if tp2_p is not None and tp3_p is not None: + reward_bands.append((tp2_p, tp3_p, REWARD_ALPHAS[2])) + elif tp3_p is not None: + reward_bands.append((tp1_p, tp3_p, REWARD_ALPHAS[2])) + elif remainder_end is not None: + reward_bands.append((tp1_p, remainder_end, REWARD_ALPHAS[1])) for price_a, price_b, alpha in reward_bands: band_low = min(price_a, price_b) band_high = max(price_a, price_b) @@ -295,6 +401,22 @@ def render_setup_chart( linewidths=0.7, ) + if ( + prev_entry_p is not None + and prev_side is not None + and prev_signal_time is not None + and prev_signal_time > 0 + ): + _draw_previous_trade( + ax, + df, + prev_entry=prev_entry_p, + prev_side=prev_side, + prev_signal_time=prev_signal_time, + exit_price=entry_p, + exit_x=entry_x, + ) + if current_p is not None: ax.hlines( current_p, @@ -321,9 +443,11 @@ def render_setup_chart( (entry_p, f"Entry {entry}", COLORS["entry"]), (sl_p, f"SL {stop_loss}", COLORS["sl"]), (tp1_p, f"TP1 {tp1}", COLORS["tp1"]), - (tp2_p, f"TP2 {tp2}", COLORS["tp2"]), - (tp3_p, f"TP3 {tp3}", COLORS["tp3"]), ] + if tp2_p is not None and tp2 is not None: + labels.append((tp2_p, f"TP2 {tp2}", COLORS["tp2"])) + if tp3_p is not None and tp3 is not None: + labels.append((tp3_p, f"TP3 {tp3}", COLORS["tp3"])) if current_p is not None and current_price is not None: labels.append((current_p, f"Price {current_price}", COLORS["price"])) for price, text, color in labels: diff --git a/app/config.py b/app/config.py index 23d31d9..8d1a7bc 100644 --- a/app/config.py +++ b/app/config.py @@ -17,6 +17,31 @@ class Settings(BaseSettings): host: str = "0.0.0.0" port: int = 8000 + scanner_enabled: bool = True + scanner_poll_seconds: float = 20.0 + watchlist_path: str = "watchlist.yaml" + scanner_state_path: str = "data/scanner.db" + risk_usd: float = 40.0 + + geryon_webhook_url: str = "" + geryon_account_id: str = "admin1" + geryon_account_id_ltf: str = "" + geryon_account_id_fvg: str = "" + geryon_order_type: str = "bbo" + geryon_secret: str = "" + geryon_secret_ltf: str = "" + geryon_secret_fvg: str = "" + + def geryon_account_id_for(self, strategy_id: str) -> str: + if strategy_id == "ltf": + return self.geryon_account_id_ltf or self.geryon_account_id + return self.geryon_account_id_fvg or self.geryon_account_id + + def geryon_secret_for(self, strategy_id: str) -> str: + if strategy_id == "ltf": + return self.geryon_secret_ltf or self.geryon_secret + return self.geryon_secret_fvg or self.geryon_secret + @lru_cache def get_settings() -> Settings: diff --git a/app/formatter.py b/app/formatter.py index a034db2..129d92d 100644 --- a/app/formatter.py +++ b/app/formatter.py @@ -101,13 +101,18 @@ def format_caption(signal: SignalPayload) -> str: signal.stop_loss_price, is_long=is_long, ) - return ( + text = ( f"{signal.ticker} {label}\n" f"\n" f"Entry price: {entry}\n" f"Price: {price}\n" f"Current profit: {profit}" ) + if signal.is_reversal and signal.realized_pnl_pct is not None: + text += ( + f"\n\nreversal, realized PnL {signal.realized_pnl_pct:+.2f}%" + ) + return text price = format_price(signal.entry_price) sl = format_price(signal.stop_loss_price) @@ -115,16 +120,30 @@ def format_caption(signal: SignalPayload) -> str: signal.entry_price, signal.stop_loss_price, is_long=is_long ) tp1 = format_price(signal.take_profit_1_price) - tp2 = format_price(signal.take_profit_2_price) - tp3 = format_price(signal.take_profit_3_price) - - return ( - f"{signal.ticker} {label}\n" - f"\n" - f"Price: {price}\n" - f"SL: {sl} {sl_pct}\n" - f"\n" - f"TP1: {tp1}\n" - f"TP2: {tp2}\n" - f"TP3: {tp3}" - ) + hold_remainder = not signal.take_profit_2_price and not signal.take_profit_3_price + lines = [ + f"{signal.ticker} {label}\n", + f"Price: {price}", + f"SL: {sl} {sl_pct}", + "", + f"TP1+BE: {tp1}" if hold_remainder else f"TP1: {tp1}", + ] + if signal.take_profit_2_price: + lines.append(f"TP2: {format_price(signal.take_profit_2_price)}") + if signal.take_profit_3_price: + lines.append(f"TP3: {format_price(signal.take_profit_3_price)}") + if hold_remainder: + lines.extend( + [ + "", + "Fix 30%, hold remainder until reversal signal.", + ] + ) + if signal.is_reversal and signal.realized_pnl_pct is not None: + lines.extend( + [ + "", + f"reversal, realized PnL {signal.realized_pnl_pct:+.2f}%", + ] + ) + return "\n".join(lines) diff --git a/app/heryon.py b/app/heryon.py new file mode 100644 index 0000000..0b0166b --- /dev/null +++ b/app/heryon.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import logging +from typing import Any + +import httpx + +from app.binance import get_tick_size +from app.config import Settings +from app.indicators.common import IndicatorSignal, calc_size, format_px, round_to_mintick + +logger = logging.getLogger(__name__) + +LTF_FIX = {"tp1_fix": "30%", "tp2_fix": "30%", "tp3_fix": "40%"} +FVG_FIX = {"tp1_fix": "30%"} + + +def to_tv_perp_ticker(symbol: str) -> str: + base = symbol.strip().upper() + if base.endswith(".P"): + return base + return f"{base}.P" + + +def heryon_nonce(ticker: str, side: str, bar_open_ts: int) -> str: + limit_side = "long" if side == "long" else "short" + return f"{ticker}-limit-{limit_side}-{bar_open_ts}" + + +def build_heryon_payload( + signal: IndicatorSignal, + *, + symbol: str, + settings: Settings, +) -> dict[str, Any]: + ticker = to_tv_perp_ticker(symbol) + tick = get_tick_size(symbol) + action = "buy" if signal.side == "long" else "sell" + fixes = LTF_FIX if signal.strategy_id == "ltf" else FVG_FIX + entry = round_to_mintick(signal.entry, tick) + sl = round_to_mintick(signal.sl, tick) + size = calc_size(entry, sl, float(settings.risk_usd)) + size_str = "" if size != size else str(int(round(size))) + payload: dict[str, Any] = { + "action": action, + "ticker": ticker, + "account_id": settings.geryon_account_id_for(signal.strategy_id), + "order_type": settings.geryon_order_type, + "position_size_usd": size_str, + "stop_loss": format_px(signal.sl, tick), + "tp1": format_px(signal.tp1, tick), + "tp1_fix": fixes.get("tp1_fix", ""), + "sl_to_bk": "tp1", + "nonce": heryon_nonce(ticker, signal.side, signal.bar_open_ts), + "secret": settings.geryon_secret_for(signal.strategy_id), + } + if signal.strategy_id == "ltf": + payload["tp2"] = format_px(signal.tp2, tick) + payload["tp2_fix"] = fixes.get("tp2_fix", "") + payload["tp3"] = format_px(signal.tp3, tick) + payload["tp3_fix"] = fixes.get("tp3_fix", "") + return {key: value for key, value in payload.items() if value != ""} + + +async def send_heryon(settings: Settings, payload: dict[str, Any]) -> None: + url = (settings.geryon_webhook_url or "").strip() + if not url: + logger.info("Heryon skipped (GERYON_WEBHOOK_URL empty): %s", payload.get("nonce")) + return + async with httpx.AsyncClient(timeout=20.0) as client: + response = await client.post(url, json=payload) + if response.status_code >= 400: + raise RuntimeError( + f"Heryon webhook {response.status_code}: {response.text[:500]}" + ) + logger.info( + "Heryon accepted nonce=%s ticker=%s action=%s", + payload.get("nonce"), + payload.get("ticker"), + payload.get("action"), + ) diff --git a/app/indicators/__init__.py b/app/indicators/__init__.py new file mode 100644 index 0000000..18600af --- /dev/null +++ b/app/indicators/__init__.py @@ -0,0 +1,11 @@ +from app.indicators.common import IndicatorSignal, calc_size, format_px +from app.indicators.fvg import evaluate_fvg +from app.indicators.ltf import evaluate_ltf + +__all__ = [ + "IndicatorSignal", + "calc_size", + "evaluate_fvg", + "evaluate_ltf", + "format_px", +] diff --git a/app/indicators/common.py b/app/indicators/common.py new file mode 100644 index 0000000..88e1e60 --- /dev/null +++ b/app/indicators/common.py @@ -0,0 +1,236 @@ +from __future__ import annotations + +from dataclasses import dataclass +from decimal import Decimal, ROUND_HALF_UP +from typing import Literal + +import numpy as np +import pandas as pd + +StrategyId = Literal["ltf", "fvg"] +Side = Literal["long", "short"] + + +@dataclass(frozen=True) +class IndicatorSignal: + strategy_id: StrategyId + side: Side + entry: float + close: float + sl: float + tp1: float + tp2: float | None + tp3: float | None + size_usd: float + bar_open_ts: int + visual_timeframe: str + + +def ohlc_frame(df: pd.DataFrame) -> pd.DataFrame: + """Normalize OHLC frame to lowercase columns and UTC DatetimeIndex.""" + out = df.copy() + out.columns = [str(c).lower() for c in out.columns] + required = {"open", "high", "low", "close"} + missing = required - set(out.columns) + if missing: + raise ValueError(f"OHLC frame missing columns: {sorted(missing)}") + if not isinstance(out.index, pd.DatetimeIndex): + raise ValueError("OHLC frame must be indexed by timestamp") + if out.index.tz is None: + out.index = out.index.tz_localize("UTC") + else: + out.index = out.index.tz_convert("UTC") + cols = ["open", "high", "low", "close"] + if "volume" in out.columns: + cols.append("volume") + return out[cols] + + +def epoch_ms(index: pd.DatetimeIndex) -> np.ndarray: + utc = index.tz_convert("UTC") if index.tz is not None else index.tz_localize("UTC") + return utc.asi8.astype(np.float64) / 1_000_000.0 + + +def alma( + series: pd.Series, length: int = 500, offset: float = 0.85, sigma: float = 5.0 +) -> pd.Series: + """Arnaud Legoux Moving Average (Pine ta.alma compatible).""" + if length < 1: + raise ValueError("ALMA length must be >= 1") + m = offset * (length - 1) + s = length / sigma + idx = np.arange(length, dtype=float) + weights = np.exp(-((idx - m) ** 2) / (2 * s * s)) + weights /= weights.sum() + + values = series.to_numpy(dtype=float) + out = np.full(len(values), np.nan, dtype=float) + if len(values) < length: + return pd.Series(out, index=series.index) + + valid = np.convolve(values, weights[::-1], mode="valid") + out[length - 1 :] = valid + nan_in_window = np.convolve(np.isnan(values).astype(float), np.ones(length), mode="valid") > 0 + out[length - 1 :][nan_in_window] = np.nan + return pd.Series(out, index=series.index) + + +def rma(series: pd.Series, length: int) -> pd.Series: + """Wilder's RMA (Pine ta.rma).""" + return series.ewm(alpha=1 / length, adjust=False, min_periods=length).mean() + + +def ha_rsi( + open_: pd.Series, + high: pd.Series, + low: pd.Series, + close: pd.Series, + length: int = 14, +) -> pd.Series: + """RSI on Heikin-Ashi close (Pine f_ha_rsi).""" + ha_close = (open_ + high + low + close) / 4.0 + delta = ha_close.diff() + up = rma(delta.clip(lower=0), length) + down = rma((-delta).clip(lower=0), length) + rs = up / down.replace(0, np.nan) + rsi = 100 - (100 / (1 + rs)) + rsi = rsi.where(down != 0, 100.0) + rsi = rsi.where(up != 0, 0.0) + both_zero = (up == 0) & (down == 0) + return rsi.where(~both_zero, 100.0) + + +def true_range(high: pd.Series, low: pd.Series, close: pd.Series) -> pd.Series: + prev_close = close.shift(1) + return pd.concat( + [high - low, (high - prev_close).abs(), (low - prev_close).abs()], + axis=1, + ).max(axis=1) + + +def atr(high: pd.Series, low: pd.Series, close: pd.Series, length: int) -> pd.Series: + """Pine ta.atr(length).""" + return rma(true_range(high, low, close), length) + + +def compute_fractals(high: np.ndarray, low: np.ndarray, n: int = 5) -> tuple[np.ndarray, np.ndarray]: + """Pine upFractal / downFractal flags on the confirmation bar (pivot at i - n).""" + size = len(high) + up = np.zeros(size, dtype=bool) + down = np.zeros(size, dtype=bool) + for c in range(n, size): + p = c - n + hp = high[p] + lp = low[p] + if np.isnan(hp) or np.isnan(lp): + continue + up_prefix1 = (p + 1 < size) and (high[p + 1] <= hp) + up_prefix2 = up_prefix1 and (p + 2 < size) and (high[p + 2] <= hp) + up_prefix3 = up_prefix2 and (p + 3 < size) and (high[p + 3] <= hp) + up_prefix4 = up_prefix3 and (p + 4 < size) and (high[p + 4] <= hp) + down_prefix1 = (p + 1 < size) and (low[p + 1] >= lp) + down_prefix2 = down_prefix1 and (p + 2 < size) and (low[p + 2] >= lp) + down_prefix3 = down_prefix2 and (p + 3 < size) and (low[p + 3] >= lp) + down_prefix4 = down_prefix3 and (p + 4 < size) and (low[p + 4] >= lp) + upflag_down = True + upflag0 = True + upflag1 = True + upflag2 = True + upflag3 = True + upflag4 = True + for i in range(1, n + 1): + if p - i < 0 or not (high[p - i] < hp): + upflag_down = False + if p + i >= size or not (high[p + i] < hp): + upflag0 = False + if p + i + 1 >= size or not (high[p + i + 1] < hp): + upflag1 = False + if p + i + 2 >= size or not (high[p + i + 2] < hp): + upflag2 = False + if p + i + 3 >= size or not (high[p + i + 3] < hp): + upflag3 = False + if p + i + 4 >= size or not (high[p + i + 4] < hp): + upflag4 = False + upflag1 = upflag1 and up_prefix1 + upflag2 = upflag2 and up_prefix2 + upflag3 = upflag3 and up_prefix3 + upflag4 = upflag4 and up_prefix4 + up[c] = upflag_down and (upflag0 or upflag1 or upflag2 or upflag3 or upflag4) + downflag_down = True + downflag0 = True + downflag1 = True + downflag2 = True + downflag3 = True + downflag4 = True + for i in range(1, n + 1): + if p - i < 0 or not (low[p - i] > lp): + downflag_down = False + if p + i >= size or not (low[p + i] > lp): + downflag0 = False + if p + i + 1 >= size or not (low[p + i + 1] > lp): + downflag1 = False + if p + i + 2 >= size or not (low[p + i + 2] > lp): + downflag2 = False + if p + i + 3 >= size or not (low[p + i + 3] > lp): + downflag3 = False + if p + i + 4 >= size or not (low[p + i + 4] > lp): + downflag4 = False + downflag1 = downflag1 and down_prefix1 + downflag2 = downflag2 and down_prefix2 + downflag3 = downflag3 and down_prefix3 + downflag4 = downflag4 and down_prefix4 + down[c] = downflag_down and (downflag0 or downflag1 or downflag2 or downflag3 or downflag4) + return up, down + + +def tick_decimals(tick: float) -> int: + exponent = Decimal(str(tick)).normalize().as_tuple().exponent + if isinstance(exponent, int): + return max(0, -exponent) + return 0 + + +def round_to_mintick(price: float, tick: float) -> float: + """Pine math.round_to_mintick (half-up to exchange tick).""" + step = Decimal(str(tick)) + if step <= 0: + return price + quantized = (Decimal(str(price)) / step).quantize(Decimal("1"), rounding=ROUND_HALF_UP) + return float(quantized * step) + + +def realized_pnl_pct(prev_side: str, prev_entry: float, exit_price: float) -> float: + """Signed percent from previous entry to exit; long/short from the closed side.""" + if prev_entry == 0: + raise ValueError("entry price is zero") + if prev_side == "long": + return (exit_price - prev_entry) / prev_entry * 100 + return (prev_entry - exit_price) / prev_entry * 100 + + +def calc_size(entry: float, sl: float, risk_usd: float) -> float: + """Pine calc_size: round(risk / abs(entry-sl)/entry).""" + if entry == 0 or np.isnan(entry) or np.isnan(sl): + return float("nan") + stop_pct = abs(entry - sl) / entry + if stop_pct <= 0: + return float("nan") + return float(round(risk_usd / stop_pct)) + + +def format_px(value: float | None, tick: float | None = None) -> str: + if value is None: + return "" + number = float(value) + if np.isnan(number): + return "" + if tick is not None and tick > 0: + rounded = round_to_mintick(number, tick) + return f"{rounded:.{tick_decimals(tick)}f}" + if abs(number) >= 1000: + text = f"{number:.4f}" + elif abs(number) >= 1: + text = f"{number:.6f}" + else: + text = f"{number:.8f}" + return text.rstrip("0").rstrip(".") diff --git a/app/indicators/fvg.py b/app/indicators/fvg.py new file mode 100644 index 0000000..89edecf --- /dev/null +++ b/app/indicators/fvg.py @@ -0,0 +1,364 @@ +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 diff --git a/app/indicators/ltf.py b/app/indicators/ltf.py new file mode 100644 index 0000000..9bfeac1 --- /dev/null +++ b/app/indicators/ltf.py @@ -0,0 +1,384 @@ +from __future__ import annotations + +from dataclasses import dataclass + +import numpy as np +import pandas as pd + +from app.indicators.common import ( + IndicatorSignal, + alma, + calc_size, + compute_fractals, + epoch_ms, + ha_rsi, + ohlc_frame, +) + + +@dataclass(frozen=True) +class LtfParams: + fractal_n: int = 5 + alma_length: int = 500 + engulf_threshold: float = 0.9 + rsi_6h_overbought: float = 75.0 + rsi_6h_oversold: float = 20.0 + tp1_rr: float = 1.75 + tp2_rr: float = 2.0 + tp3_rr: float = 4.0 + risk_usd: float = 40.0 + cooldown_ms: int = 360 * 60 * 1000 + engulf_reset_ms: int = 12 * 60 * 60 * 1000 + tf_minutes: int = 15 + visual_timeframe: str = "15" + + +def _merge_asof(df_15m: pd.DataFrame, df_6h: pd.DataFrame, columns: list[str]) -> pd.DataFrame: + left = pd.DataFrame({"ts": df_15m.index}) + right = pd.DataFrame({"ts": df_6h.index}) + for col in columns: + right[col] = df_6h[col].to_numpy() + left = left.sort_values("ts") + right = right.sort_values("ts") + merged = pd.merge_asof(left, right, on="ts", direction="backward") + merged = merged.set_index("ts") + return merged.reindex(df_15m.index) + + +def _compute_6h( + df_6h: pd.DataFrame, + *, + engulf_threshold: float, + forming: pd.Series | None, +) -> pd.DataFrame: + df = ohlc_frame(df_6h) + df["ha_rsi"] = ha_rsi(df["open"], df["high"], df["low"], df["close"], 14) + df["h6_close"] = df["close"] + df["h6_prev_close"] = df["close"].shift(1) + df["h6_open"] = df["open"] + df["h6_prev_open"] = df["open"].shift(1) + + body2 = (df["h6_prev_open"] - df["h6_prev_close"]).abs() + bull_min = df["h6_prev_close"] + body2 * engulf_threshold + bear_min = df["h6_prev_close"] - body2 * engulf_threshold + bull_engulf = ( + (df["h6_close"] > df["h6_open"]) + & (df["h6_close"] >= bull_min) + & (df["h6_prev_close"] < df["h6_prev_open"]) + ) + bear_engulf = ( + (df["h6_close"] < df["h6_open"]) + & (df["h6_close"] <= bear_min) + & (df["h6_prev_close"] > df["h6_prev_open"]) + ) + + n = len(df) + is_forming = np.zeros(n, dtype=bool) + if forming is not None: + is_forming = forming.reindex(df.index).fillna(False).to_numpy(dtype=bool) + + state = np.empty(n, dtype=object) + state[:] = "" + engulf_time = np.full(n, np.nan) + cur_state = "" + cur_time = np.nan + ts_ms = epoch_ms(df.index) + bull_a = bull_engulf.fillna(False).to_numpy(dtype=bool) + bear_a = bear_engulf.fillna(False).to_numpy(dtype=bool) + + for i in range(n): + ts = ts_ms[i] + if not np.isnan(cur_time) and (ts - cur_time) > 12 * 60 * 60 * 1000: + cur_state = "" + cur_time = np.nan + + if is_forming[i]: + state[i] = cur_state + engulf_time[i] = cur_time + continue + + if bull_a[i]: + cur_state = "bull" + cur_time = ts + elif bear_a[i]: + cur_state = "bear" + cur_time = ts + else: + cur_state = "" + cur_time = np.nan + + state[i] = cur_state + engulf_time[i] = cur_time + + df["engulf_state"] = state + df["engulf_time"] = engulf_time + return df + + +def attach_forming_6h(df_6h_closed: pd.DataFrame, df_15m: pd.DataFrame) -> pd.DataFrame: + """Rebuild the in-progress 6h candle from closed 15m bars (Pine request.security).""" + closed = ohlc_frame(df_6h_closed) + ltf = ohlc_frame(df_15m) + if ltf.empty: + closed["_forming"] = False + return closed + + period = ltf.index[-1].floor("6h") + window = ltf.loc[ltf.index >= period] + if window.empty: + closed["_forming"] = False + return closed + + # 6h just closed and is already in the closed frame. + if not closed.empty and closed.index[-1] == period: + closed["_forming"] = False + return closed + + row = pd.DataFrame( + { + "open": [float(window["open"].iloc[0])], + "high": [float(window["high"].max())], + "low": [float(window["low"].min())], + "close": [float(window["close"].iloc[-1])], + }, + index=pd.DatetimeIndex([period], tz="UTC"), + ) + if "volume" in window.columns: + row["volume"] = float(window["volume"].sum()) + row["_forming"] = True + base = closed[closed.index < period].copy() + base["_forming"] = False + return pd.concat([base, row]) + + +def evaluate_ltf( + df_15m: pd.DataFrame, + df_6h: pd.DataFrame, + params: LtfParams | None = None, + *, + forming_6h: pd.Series | None = None, +) -> pd.DataFrame: + """Compute LTF columns. Last row is the latest closed 15m bar.""" + p = params or LtfParams() + df = ohlc_frame(df_15m) + forming_flag = forming_6h + if forming_flag is None and "_forming" in df_6h.columns: + forming_flag = df_6h["_forming"].astype(bool) + h6 = _compute_6h(df_6h, engulf_threshold=p.engulf_threshold, forming=forming_flag) + if forming_flag is not None: + closed_mask = ~forming_flag.reindex(h6.index).fillna(False) + h6_closed = h6.loc[closed_mask] + else: + h6_closed = h6 + + n = int(p.fractal_n) + high = df["high"].to_numpy(dtype=float) + low = df["low"].to_numpy(dtype=float) + up_frac, down_frac = compute_fractals(high, low, n=n) + df["up_fractal"] = up_frac + df["down_fractal"] = down_frac + + up_level = np.full(len(df), np.nan) + down_level = np.full(len(df), np.nan) + last_up = np.nan + last_down = np.nan + for i in range(len(df)): + if up_frac[i]: + last_up = high[i - n] + if down_frac[i]: + last_down = low[i - n] + up_level[i] = last_up + down_level[i] = last_down + df["up_fractal_level"] = up_level + df["down_fractal_level"] = down_level + + prev_up = df["up_fractal_level"].shift(1) + prev_down = df["down_fractal_level"].shift(1) + df["buy_crossover"] = ( + (df["close"] > prev_up) & (df["close"].shift(1) <= prev_up.shift(1)) & prev_up.notna() + ) + df["sell_crossover"] = ( + (df["close"] < prev_down) & (df["close"].shift(1) >= prev_down.shift(1)) & prev_down.notna() + ) + + bars_16h = max(1, int(round(1440 / p.tf_minutes))) + high_shift = df["high"].shift(1) + low_shift = df["low"].shift(1) + df["high_16h"] = high_shift.rolling(bars_16h, min_periods=bars_16h).max() + df["low_16h"] = low_shift.rolling(bars_16h, min_periods=bars_16h).min() + df["bull_breakout"] = df["high"] > df["high_16h"] + df["bear_breakout"] = df["low"] < df["low_16h"] + + df["alma"] = alma(df["close"], length=p.alma_length, offset=0.85, sigma=5.0) + + rsi_m = _merge_asof(df, h6, ["ha_rsi"]) + closed_src = h6_closed if not h6_closed.empty else h6 + closed_m = _merge_asof( + df, closed_src, ["engulf_state", "engulf_time", "h6_close", "h6_prev_close"] + ) + df["rsi_6h"] = rsi_m["ha_rsi"] + df["h6_engulf_state"] = closed_m["engulf_state"].fillna("").astype(str) + df["h6_closed_close"] = closed_m["h6_close"] + df["h6_prev_close"] = closed_m["h6_prev_close"] + engulf_time = closed_m["engulf_time"].to_numpy(dtype=float) + + ts_ms = epoch_ms(df.index) + engulf_state = df["h6_engulf_state"].to_numpy() + for i in range(len(df)): + et = engulf_time[i] + if not np.isnan(et) and (ts_ms[i] - et) > p.engulf_reset_ms: + engulf_state[i] = "" + df["h6_engulf_state"] = engulf_state + + bull_arr = (df["bull_breakout"] | (df["h6_engulf_state"] == "bull")).to_numpy(dtype=bool) + bear_arr = (df["bear_breakout"] | (df["h6_engulf_state"] == "bear")).to_numpy(dtype=bool) + last_eng = np.empty(len(df), dtype=object) + last_eng[:] = "" + cur = "" + for i in range(len(df)): + b = bull_arr[i] + s = bear_arr[i] + if b and not s: + cur = "bull" + elif s and not b: + cur = "bear" + last_eng[i] = cur + df["last_eng"] = last_eng + + alma_up = (df["alma"] > df["alma"].shift(1)) & (df["alma"].shift(1) > df["alma"].shift(2)) + alma_down = (df["alma"] < df["alma"].shift(1)) & (df["alma"].shift(1) < df["alma"].shift(2)) + alma_6h_long = (df["h6_closed_close"] > df["alma"]) & ( + df["h6_prev_close"] > df["alma"].shift(1) + ) + alma_6h_short = (df["h6_closed_close"] < df["alma"]) & ( + df["h6_prev_close"] < df["alma"].shift(1) + ) + buy_6h_ok = df["rsi_6h"] < p.rsi_6h_overbought + sell_6h_ok = df["rsi_6h"] > p.rsi_6h_oversold + + buy_sig = np.zeros(len(df), dtype=bool) + sell_sig = np.zeros(len(df), dtype=bool) + last_buy_time = np.nan + last_sell_time = np.nan + open_side = "" + long_sl_px = np.nan + short_sl_px = np.nan + close_a = df["close"].to_numpy(dtype=float) + low_a = df["low"].to_numpy(dtype=float) + high_a = df["high"].to_numpy(dtype=float) + buy_x = df["buy_crossover"].fillna(False).to_numpy(dtype=bool) + sell_x = df["sell_crossover"].fillna(False).to_numpy(dtype=bool) + alma_up_a = alma_up.fillna(False).to_numpy(dtype=bool) + alma_down_a = alma_down.fillna(False).to_numpy(dtype=bool) + alma_6h_long_a = alma_6h_long.fillna(False).to_numpy(dtype=bool) + alma_6h_short_a = alma_6h_short.fillna(False).to_numpy(dtype=bool) + buy_6h_a = buy_6h_ok.fillna(False).to_numpy(dtype=bool) + sell_6h_a = sell_6h_ok.fillna(False).to_numpy(dtype=bool) + long_sl_arr = df["low_16h"].to_numpy(dtype=float) + short_sl_arr = df["high_16h"].to_numpy(dtype=float) + + for i in range(len(df)): + if not np.isnan(long_sl_px) and (close_a[i] < long_sl_px or low_a[i] <= long_sl_px): + long_sl_px = np.nan + if open_side == "buy": + open_side = "" + if not np.isnan(short_sl_px) and (close_a[i] > short_sl_px or high_a[i] >= short_sl_px): + short_sl_px = np.nan + if open_side == "sell": + open_side = "" + + ts = ts_ms[i] + buy_cd = np.isnan(last_buy_time) or (ts - last_buy_time >= p.cooldown_ms) + sell_cd = np.isnan(last_sell_time) or (ts - last_sell_time >= p.cooldown_ms) + + is_buy = ( + buy_x[i] + and open_side != "buy" + and last_eng[i] == "bull" + and (alma_up_a[i] or alma_6h_long_a[i]) + and buy_cd + and buy_6h_a[i] + ) + is_sell = ( + sell_x[i] + and open_side != "sell" + and last_eng[i] == "bear" + and (alma_down_a[i] or alma_6h_short_a[i]) + and sell_cd + and sell_6h_a[i] + ) + + if is_buy: + buy_sig[i] = True + last_buy_time = ts + open_side = "buy" + long_sl_px = long_sl_arr[i] + short_sl_px = np.nan + if is_sell: + sell_sig[i] = True + last_sell_time = ts + open_side = "sell" + short_sl_px = short_sl_arr[i] + long_sl_px = np.nan + + df["buy_sig"] = buy_sig + df["sell_sig"] = sell_sig + df["long_sl"] = df["low_16h"] + df["short_sl"] = df["high_16h"] + df["long_tp1"] = df["close"] + (df["close"] - df["long_sl"]) * p.tp1_rr + df["long_tp2"] = df["close"] + (df["close"] - df["long_sl"]) * p.tp2_rr + df["long_tp3"] = df["close"] + (df["close"] - df["long_sl"]) * p.tp3_rr + df["short_tp1"] = df["close"] - (df["short_sl"] - df["close"]) * p.tp1_rr + df["short_tp2"] = df["close"] - (df["short_sl"] - df["close"]) * p.tp2_rr + df["short_tp3"] = df["close"] - (df["short_sl"] - df["close"]) * p.tp3_rr + return df + + +def last_ltf_signal(df: pd.DataFrame, params: LtfParams | None = None) -> IndicatorSignal | None: + if df.empty: + return None + p = params or LtfParams() + last = df.iloc[-1] + ts = int(df.index[-1].timestamp()) + close = float(last["close"]) + if bool(last["buy_sig"]): + sl = float(last["long_sl"]) + if np.isnan(sl) or np.isnan(close): + return None + size = calc_size(close, sl, p.risk_usd) + if np.isnan(size): + return None + return IndicatorSignal( + strategy_id="ltf", + side="long", + entry=close, + close=close, + sl=sl, + tp1=float(last["long_tp1"]), + tp2=float(last["long_tp2"]), + tp3=float(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"]) + if np.isnan(sl) or np.isnan(close): + return None + size = calc_size(close, sl, p.risk_usd) + if np.isnan(size): + return None + return IndicatorSignal( + strategy_id="ltf", + side="short", + entry=close, + close=close, + sl=sl, + tp1=float(last["short_tp1"]), + tp2=float(last["short_tp2"]), + tp3=float(last["short_tp3"]), + size_usd=size, + bar_open_ts=ts, + visual_timeframe=p.visual_timeframe, + ) + return None diff --git a/app/main.py b/app/main.py index 192aa92..b60f0e2 100644 --- a/app/main.py +++ b/app/main.py @@ -4,18 +4,16 @@ import asyncio import json import logging import secrets +from contextlib import asynccontextmanager from typing import Any from fastapi import BackgroundTasks, FastAPI, HTTPException, Request from fastapi.responses import JSONResponse from pydantic import ValidationError -from app.binance import fetch_klines, to_binance_interval, to_binance_symbol -from app.chart import render_setup_chart from app.config import Settings, get_settings -from app.formatter import format_caption from app.models import SignalPayload -from app.telegram import TelegramError, send_message, send_photo +from app.pipeline import deliver_telegram logging.basicConfig( level=logging.INFO, @@ -23,12 +21,37 @@ logging.basicConfig( ) logger = logging.getLogger(__name__) + +@asynccontextmanager +async def lifespan(_app: FastAPI): + settings = get_settings() + task: asyncio.Task[None] | None = None + if settings.scanner_enabled: + from app.scanner import run_scanner + + task = asyncio.create_task(run_scanner(settings)) + logger.info( + "Scanner enabled watchlist=%s poll=%ss", + settings.watchlist_path, + settings.scanner_poll_seconds, + ) + yield + if task is not None: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + logger.info("Scanner stopped") + + app = FastAPI( title="TV Signals → Telegram", version="1.0.0", docs_url=None, redoc_url=None, openapi_url=None, + lifespan=lifespan, ) @@ -53,76 +76,6 @@ def _parse_signal_body(raw: bytes) -> SignalPayload: raise HTTPException(status_code=422, detail=json.loads(exc.json())) from exc -async def _deliver_signal( - settings: Settings, - signal: SignalPayload, - message_thread_id: int, -) -> None: - """Fetch chart + post to Telegram after the webhook HTTP response is sent.""" - caption = format_caption(signal) - photo: bytes | None = None - chart_error: str | None = None - - try: - symbol = to_binance_symbol(signal.ticker) - interval = to_binance_interval(signal.visual_timeframe) - df = await fetch_klines(symbol, interval) - photo = await asyncio.to_thread( - render_setup_chart, - df, - ticker=signal.ticker, - action=signal.action.value, - entry=signal.entry_price, - stop_loss=signal.stop_loss_price, - tp1=signal.take_profit_1_price, - tp2=signal.take_profit_2_price, - tp3=signal.take_profit_3_price, - timeframe=signal.visual_timeframe, - current_price=( - signal.current_price if signal.signal_sequence > 1 else None - ), - signal_time=signal.signal_time, - ) - except Exception as exc: # noqa: BLE001 — fallback to text-only post - chart_error = str(exc) - logger.exception("Chart generation failed, falling back to text-only: %s", exc) - - try: - if photo is not None: - await send_photo( - settings, - photo=photo, - caption=caption, - message_thread_id=message_thread_id, - ) - logger.info( - "Delivered photo: %s %s seq=%s thread=%s", - signal.ticker, - signal.action.value, - signal.signal_sequence, - message_thread_id, - ) - return - - await send_message( - settings, - text=caption, - message_thread_id=message_thread_id, - ) - logger.info( - "Delivered text: %s %s seq=%s thread=%s chart_error=%s", - signal.ticker, - signal.action.value, - signal.signal_sequence, - message_thread_id, - chart_error, - ) - except TelegramError as exc: - logger.exception("Telegram delivery failed: %s", exc) - except Exception as exc: # noqa: BLE001 - logger.exception("Unexpected delivery error: %s", exc) - - @app.api_route("/health", methods=["GET", "HEAD"]) async def health() -> dict[str, str]: return {"status": "ok"} @@ -153,7 +106,7 @@ async def _handle_webhook( ) background_tasks.add_task( - _deliver_signal, + deliver_telegram, settings, signal, message_thread_id, diff --git a/app/models.py b/app/models.py index c8bb966..25e065f 100644 --- a/app/models.py +++ b/app/models.py @@ -1,4 +1,7 @@ +from __future__ import annotations + from enum import Enum +from typing import Optional from pydantic import BaseModel, Field, field_validator @@ -15,11 +18,16 @@ class SignalPayload(BaseModel): current_price: str stop_loss_price: str take_profit_1_price: str - take_profit_2_price: str - take_profit_3_price: str + take_profit_2_price: Optional[str] = None + take_profit_3_price: Optional[str] = None visual_timeframe: str signal_sequence: int = Field(ge=1) signal_time: int = Field(gt=0) # unix seconds of seq==1 bar open (UTC) + is_reversal: bool = False + realized_pnl_pct: Optional[float] = None + prev_side: Optional[str] = None + prev_entry_price: Optional[str] = None + prev_signal_time: Optional[int] = None @field_validator("action", mode="before") @classmethod @@ -34,8 +42,6 @@ class SignalPayload(BaseModel): "current_price", "stop_loss_price", "take_profit_1_price", - "take_profit_2_price", - "take_profit_3_price", "visual_timeframe", mode="before", ) @@ -44,3 +50,14 @@ class SignalPayload(BaseModel): if isinstance(value, str): return value.strip() return value + + @field_validator("take_profit_2_price", "take_profit_3_price", mode="before") + @classmethod + def empty_optional_price(cls, value: object) -> object: + if value is None: + return None + if isinstance(value, str) and not value.strip(): + return None + if isinstance(value, str): + return value.strip() + return value diff --git a/app/pipeline.py b/app/pipeline.py new file mode 100644 index 0000000..0baf0ad --- /dev/null +++ b/app/pipeline.py @@ -0,0 +1,181 @@ +from __future__ import annotations + +import asyncio +import logging + +from app.binance import ( + chart_kline_limit, + chart_right_pad, + fetch_klines, + get_tick_size, + interval_timedelta, + to_binance_interval, + to_binance_symbol, +) +from app.chart import render_setup_chart +from app.config import Settings +from app.formatter import format_caption +from app.heryon import build_heryon_payload, send_heryon, to_tv_perp_ticker +from app.indicators.common import IndicatorSignal, format_px, realized_pnl_pct +from app.models import SignalPayload +from app.state import ScannerStore +from app.telegram import TelegramError, send_message, send_photo, telegram_message_id + +logger = logging.getLogger(__name__) + + +def to_telegram_payload(signal: IndicatorSignal, ticker: str) -> SignalPayload: + tick = get_tick_size(ticker) + return SignalPayload( + ticker=ticker, + action=signal.side, + entry_price=format_px(signal.entry, tick), + current_price=format_px(signal.close, tick), + stop_loss_price=format_px(signal.sl, tick), + take_profit_1_price=format_px(signal.tp1, tick), + take_profit_2_price=format_px(signal.tp2, tick) or None, + take_profit_3_price=format_px(signal.tp3, tick) or None, + visual_timeframe=signal.visual_timeframe, + signal_sequence=1, + signal_time=signal.bar_open_ts, + ) + + +async def deliver_telegram( + settings: Settings, + signal: SignalPayload, + message_thread_id: int, + *, + store: ScannerStore | None = None, +) -> None: + """Fetch chart + post to Telegram (used by inbound webhook and scanner).""" + caption = format_caption(signal) + photo: bytes | None = None + chart_error: str | None = None + owns_store = store is None + db = store or ScannerStore(settings.scanner_state_path) + reply_id = db.get_last_tg_message(message_thread_id, signal.ticker) + + try: + symbol = to_binance_symbol(signal.ticker) + interval = to_binance_interval(signal.visual_timeframe) + limit = chart_kline_limit(interval) + if signal.prev_signal_time and signal.prev_signal_time > 0: + bar_sec = interval_timedelta(interval).total_seconds() + if bar_sec > 0: + span = int((signal.signal_time - signal.prev_signal_time) / bar_sec) + 24 + limit = min(1500, max(limit, span)) + df = await fetch_klines(symbol, interval, limit=limit) + # LTF 15m: chart shows TP1 + remainder green (like FVG); caption still has TP2/TP3. + chart_tp2 = None if interval == "15m" else signal.take_profit_2_price + chart_tp3 = None if interval == "15m" else signal.take_profit_3_price + photo = await asyncio.to_thread( + render_setup_chart, + df, + ticker=signal.ticker, + action=signal.action.value, + entry=signal.entry_price, + stop_loss=signal.stop_loss_price, + tp1=signal.take_profit_1_price, + tp2=chart_tp2, + tp3=chart_tp3, + timeframe=signal.visual_timeframe, + current_price=( + signal.current_price if signal.signal_sequence > 1 else None + ), + signal_time=signal.signal_time, + right_pad=chart_right_pad(interval), + prev_entry=signal.prev_entry_price, + prev_side=signal.prev_side, + prev_signal_time=signal.prev_signal_time, + ) + except Exception as exc: # noqa: BLE001 — fallback to text-only post + chart_error = str(exc) + logger.exception("Chart generation failed, falling back to text-only: %s", exc) + + try: + if photo is not None: + response = await send_photo( + settings, + photo=photo, + caption=caption, + message_thread_id=message_thread_id, + reply_to_message_id=reply_id, + ) + else: + response = await send_message( + settings, + text=caption, + message_thread_id=message_thread_id, + reply_to_message_id=reply_id, + ) + message_id = telegram_message_id(response) + if message_id is not None: + db.set_last_tg_message(message_thread_id, signal.ticker, message_id) + logger.info( + "Delivered %s: %s %s seq=%s thread=%s chart_error=%s", + "photo" if photo is not None else "text", + signal.ticker, + signal.action.value, + signal.signal_sequence, + message_thread_id, + chart_error, + ) + except TelegramError as exc: + logger.exception("Telegram delivery failed: %s", exc) + except Exception as exc: # noqa: BLE001 + logger.exception("Unexpected delivery error: %s", exc) + finally: + if owns_store: + db.close() + + +async def deliver_generated( + settings: Settings, + signal: IndicatorSignal, + symbol: str, + message_thread_id: int, + *, + store: ScannerStore | None = None, +) -> None: + ticker = to_tv_perp_ticker(symbol) + telegram_payload = to_telegram_payload(signal, ticker) + if store is not None: + previous = store.get_last_open(signal.strategy_id, symbol) + if previous is not None: + prev_side, prev_entry, prev_ts = previous + if prev_side != signal.side: + tick = get_tick_size(ticker) + telegram_payload = telegram_payload.model_copy( + update={ + "is_reversal": True, + "realized_pnl_pct": realized_pnl_pct( + prev_side, prev_entry, signal.entry + ), + "prev_side": prev_side, + "prev_entry_price": format_px(prev_entry, tick), + "prev_signal_time": prev_ts or None, + } + ) + heryon_payload = build_heryon_payload(signal, symbol=symbol, settings=settings) + + await deliver_telegram( + settings, + telegram_payload, + message_thread_id, + store=store, + ) + if store is not None: + store.set_last_open( + signal.strategy_id, + symbol, + signal.side, + signal.entry, + signal.bar_open_ts, + ) + try: + await send_heryon(settings, heryon_payload) + except Exception as exc: # noqa: BLE001 + logger.exception( + "Heryon send failed nonce=%s: %s", heryon_payload.get("nonce"), exc + ) diff --git a/app/scanner.py b/app/scanner.py new file mode 100644 index 0000000..8e6d12b --- /dev/null +++ b/app/scanner.py @@ -0,0 +1,164 @@ +from __future__ import annotations + +import asyncio +import logging + +import httpx + +from app.binance import fetch_klines, load_tick_sizes, to_binance_symbol +from app.config import Settings +from app.heryon import heryon_nonce, to_tv_perp_ticker +from app.indicators.fvg import FvgParams, evaluate_fvg, last_fvg_signal +from app.indicators.ltf import LtfParams, attach_forming_6h, evaluate_ltf, last_ltf_signal +from app.pipeline import deliver_generated +from app.state import ScannerStore +from app.watchlist import Watchlist, load_watchlist + +logger = logging.getLogger(__name__) + +LTF_15M_BARS = 600 +LTF_6H_BARS = 250 +FVG_6H_BARS = 400 + + +async def run_scanner(settings: Settings) -> None: + store = ScannerStore(settings.scanner_state_path) + try: + await load_tick_sizes() + while True: + try: + watchlist = load_watchlist(settings.watchlist_path) + await scan_once(settings, watchlist, store) + except asyncio.CancelledError: + raise + except Exception: + logger.exception("Scanner cycle failed") + await asyncio.sleep(settings.scanner_poll_seconds) + finally: + store.close() + + +async def scan_once( + settings: Settings, + watchlist: Watchlist, + store: ScannerStore, +) -> None: + ltf_cfg = watchlist.strategies.get("ltf") + fvg_cfg = watchlist.strategies.get("fvg") + async with httpx.AsyncClient(timeout=20.0) as client: + for symbol in watchlist.symbols: + pair = to_binance_symbol(symbol) + if ltf_cfg is not None and ltf_cfg.enabled: + try: + await _scan_ltf(settings, store, client, pair, ltf_cfg.telegram_thread_id) + except Exception: + logger.exception("LTF scan failed for %s", pair) + if fvg_cfg is not None and fvg_cfg.enabled: + try: + await _scan_fvg(settings, store, client, pair, fvg_cfg.telegram_thread_id) + except Exception: + logger.exception("FVG scan failed for %s", pair) + + +def _thread(settings: Settings, override: int | None) -> int: + if override is not None and override >= 1: + return override + return settings.telegram_message_thread_id + + +async def _scan_ltf( + settings: Settings, + store: ScannerStore, + client: httpx.AsyncClient, + symbol: str, + thread_id: int | None, +) -> None: + df_15m = await fetch_klines( + symbol, "15m", limit=LTF_15M_BARS, closed_only=True, client=client + ) + if df_15m.empty: + return + bar_ts = int(df_15m.index[-1].timestamp()) + last = store.get_last_bar("ltf", symbol) + if last is None: + store.set_last_bar("ltf", symbol, bar_ts) + logger.info("LTF primed %s at %s (skip history)", symbol, bar_ts) + return + if bar_ts <= last: + return + + df_6h_closed = await fetch_klines( + symbol, "6h", limit=LTF_6H_BARS, closed_only=True, client=client + ) + if df_6h_closed.empty: + logger.warning("LTF %s: empty 6h klines", symbol) + return + df_6h = attach_forming_6h(df_6h_closed, df_15m) + params = LtfParams(risk_usd=settings.risk_usd) + analyzed = await asyncio.to_thread(evaluate_ltf, df_15m, df_6h, params) + signal = last_ltf_signal(analyzed, params) + await _emit_if_new(settings, store, "ltf", symbol, bar_ts, signal, thread_id) + + +async def _scan_fvg( + settings: Settings, + store: ScannerStore, + client: httpx.AsyncClient, + symbol: str, + thread_id: int | None, +) -> None: + df_6h = await fetch_klines( + symbol, "6h", limit=FVG_6H_BARS, closed_only=True, client=client + ) + if df_6h.empty: + return + bar_ts = int(df_6h.index[-1].timestamp()) + last = store.get_last_bar("fvg", symbol) + if last is None: + store.set_last_bar("fvg", symbol, bar_ts) + logger.info("FVG primed %s at %s (skip history)", symbol, bar_ts) + return + if bar_ts <= last: + return + + params = FvgParams(risk_usd=settings.risk_usd) + analyzed = await asyncio.to_thread(evaluate_fvg, df_6h, params) + signal = last_fvg_signal(analyzed, params) + await _emit_if_new(settings, store, "fvg", symbol, bar_ts, signal, thread_id) + + +async def _emit_if_new( + settings: Settings, + store: ScannerStore, + strategy: str, + symbol: str, + bar_ts: int, + signal, + thread_id: int | None, +) -> None: + if signal is None: + store.set_last_bar(strategy, symbol, bar_ts) + logger.debug("%s %s new bar %s — no signal", strategy, symbol, bar_ts) + return + + ticker = to_tv_perp_ticker(symbol) + nonce = heryon_nonce(ticker, signal.side, signal.bar_open_ts) + if store.nonce_sent(nonce): + store.set_last_bar(strategy, symbol, bar_ts) + logger.info("Skip duplicate nonce %s", nonce) + return + + logger.info( + "Signal %s %s %s bar=%s sl=%s tp1=%s", + strategy, + symbol, + signal.side, + bar_ts, + signal.sl, + signal.tp1, + ) + await deliver_generated( + settings, signal, symbol, _thread(settings, thread_id), store=store + ) + store.mark_nonce(nonce) + store.set_last_bar(strategy, symbol, bar_ts) diff --git a/app/state.py b/app/state.py new file mode 100644 index 0000000..9b01463 --- /dev/null +++ b/app/state.py @@ -0,0 +1,145 @@ +from __future__ import annotations + +import sqlite3 +import time +from pathlib import Path + + +class ScannerStore: + """Persist last processed bar per (strategy, symbol) and sent Heryon nonces.""" + + def __init__(self, path: str | Path) -> None: + self._path = Path(path) + self._path.parent.mkdir(parents=True, exist_ok=True) + self._conn = sqlite3.connect(self._path, check_same_thread=False) + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS processed ( + strategy TEXT NOT NULL, + symbol TEXT NOT NULL, + bar_open_ts INTEGER NOT NULL, + PRIMARY KEY (strategy, symbol) + ) + """ + ) + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS sent_nonces ( + nonce TEXT PRIMARY KEY, + sent_at INTEGER NOT NULL + ) + """ + ) + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS last_open ( + strategy TEXT NOT NULL, + symbol TEXT NOT NULL, + side TEXT NOT NULL, + entry REAL NOT NULL, + bar_open_ts INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (strategy, symbol) + ) + """ + ) + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS last_tg_message ( + thread_id INTEGER NOT NULL, + ticker TEXT NOT NULL, + message_id INTEGER NOT NULL, + PRIMARY KEY (thread_id, ticker) + ) + """ + ) + last_open_cols = { + row[1] for row in self._conn.execute("PRAGMA table_info(last_open)") + } + if "bar_open_ts" not in last_open_cols: + self._conn.execute( + "ALTER TABLE last_open ADD COLUMN bar_open_ts INTEGER NOT NULL DEFAULT 0" + ) + self._conn.commit() + + def get_last_bar(self, strategy: str, symbol: str) -> int | None: + row = self._conn.execute( + "SELECT bar_open_ts FROM processed WHERE strategy = ? AND symbol = ?", + (strategy, symbol), + ).fetchone() + return int(row[0]) if row else None + + def set_last_bar(self, strategy: str, symbol: str, bar_open_ts: int) -> None: + self._conn.execute( + """ + INSERT INTO processed (strategy, symbol, bar_open_ts) + VALUES (?, ?, ?) + ON CONFLICT(strategy, symbol) DO UPDATE SET bar_open_ts = excluded.bar_open_ts + """, + (strategy, symbol, bar_open_ts), + ) + self._conn.commit() + + def nonce_sent(self, nonce: str) -> bool: + row = self._conn.execute( + "SELECT 1 FROM sent_nonces WHERE nonce = ?", (nonce,) + ).fetchone() + return row is not None + + def mark_nonce(self, nonce: str) -> None: + self._conn.execute( + "INSERT OR IGNORE INTO sent_nonces (nonce, sent_at) VALUES (?, ?)", + (nonce, int(time.time())), + ) + self._conn.commit() + + def get_last_open(self, strategy: str, symbol: str) -> tuple[str, float, int] | None: + row = self._conn.execute( + "SELECT side, entry, bar_open_ts FROM last_open WHERE strategy = ? AND symbol = ?", + (strategy, symbol), + ).fetchone() + if row is None: + return None + return str(row[0]), float(row[1]), int(row[2] or 0) + + def set_last_open( + self, + strategy: str, + symbol: str, + side: str, + entry: float, + bar_open_ts: int, + ) -> None: + self._conn.execute( + """ + INSERT INTO last_open (strategy, symbol, side, entry, bar_open_ts) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(strategy, symbol) DO UPDATE SET + side = excluded.side, + entry = excluded.entry, + bar_open_ts = excluded.bar_open_ts + """, + (strategy, symbol, side, entry, bar_open_ts), + ) + self._conn.commit() + + def get_last_tg_message(self, thread_id: int, ticker: str) -> int | None: + row = self._conn.execute( + "SELECT message_id FROM last_tg_message WHERE thread_id = ? AND ticker = ?", + (thread_id, ticker.strip().upper()), + ).fetchone() + return int(row[0]) if row else None + + def set_last_tg_message(self, thread_id: int, ticker: str, message_id: int) -> None: + self._conn.execute( + """ + INSERT INTO last_tg_message (thread_id, ticker, message_id) + VALUES (?, ?, ?) + ON CONFLICT(thread_id, ticker) DO UPDATE SET message_id = excluded.message_id + """, + (thread_id, ticker.strip().upper(), message_id), + ) + self._conn.commit() + + def close(self) -> None: + self._conn.close() diff --git a/app/telegram.py b/app/telegram.py index 3a073b2..5dce001 100644 --- a/app/telegram.py +++ b/app/telegram.py @@ -15,6 +15,23 @@ class TelegramError(Exception): pass +def telegram_message_id(payload: dict) -> int | None: + result = payload.get("result") + if not isinstance(result, dict): + return None + raw = result.get("message_id") + return int(raw) if raw is not None else None + + +def _reply_fields(reply_to_message_id: int | None) -> dict[str, str]: + if reply_to_message_id is None: + return {} + return { + "reply_to_message_id": str(reply_to_message_id), + "allow_sending_without_reply": "true", + } + + async def send_photo( settings: Settings, *, @@ -22,6 +39,7 @@ async def send_photo( caption: str, message_thread_id: int, filename: str = "setup.png", + reply_to_message_id: int | None = None, ) -> dict: url = f"{TELEGRAM_API}/bot{settings.telegram_bot_token}/sendPhoto" data = { @@ -29,6 +47,7 @@ async def send_photo( "message_thread_id": str(message_thread_id), "caption": caption, "parse_mode": "HTML", + **_reply_fields(reply_to_message_id), } files = {"photo": (filename, photo, "image/png")} async with httpx.AsyncClient(timeout=30.0) as client: @@ -44,6 +63,7 @@ async def send_message( *, text: str, message_thread_id: int, + reply_to_message_id: int | None = None, ) -> dict: url = f"{TELEGRAM_API}/bot{settings.telegram_bot_token}/sendMessage" data = { @@ -52,6 +72,7 @@ async def send_message( "text": text, "parse_mode": "HTML", "disable_web_page_preview": True, + **_reply_fields(reply_to_message_id), } async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post(url, data=data) diff --git a/app/watchlist.py b/app/watchlist.py new file mode 100644 index 0000000..56eb76c --- /dev/null +++ b/app/watchlist.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import yaml + + +@dataclass(frozen=True) +class StrategyWatch: + id: str + enabled: bool + timeframe: str + telegram_thread_id: int | None + + +@dataclass(frozen=True) +class Watchlist: + symbols: list[str] + strategies: dict[str, StrategyWatch] + + +def load_watchlist(path: str | Path) -> Watchlist: + raw_path = Path(path) + if not raw_path.is_file(): + raise FileNotFoundError(f"Watchlist not found: {raw_path}") + data: Any = yaml.safe_load(raw_path.read_text(encoding="utf-8")) or {} + symbols_raw = data.get("symbols") or [] + symbols = [str(s).strip().upper() for s in symbols_raw if str(s).strip()] + if not symbols: + raise ValueError("Watchlist has no symbols") + + strategies: dict[str, StrategyWatch] = {} + for key, cfg in (data.get("strategies") or {}).items(): + if not isinstance(cfg, dict): + continue + strategies[str(key)] = StrategyWatch( + id=str(key), + enabled=bool(cfg.get("enabled", True)), + timeframe=str(cfg.get("timeframe", "")).strip(), + telegram_thread_id=_optional_int(cfg.get("telegram_thread_id")), + ) + if "ltf" not in strategies: + strategies["ltf"] = StrategyWatch("ltf", True, "15m", None) + if "fvg" not in strategies: + strategies["fvg"] = StrategyWatch("fvg", True, "6h", None) + return Watchlist(symbols=symbols, strategies=strategies) + + +def _optional_int(value: object) -> int | None: + if value is None or value == "": + return None + return int(value) diff --git a/data/.gitkeep b/data/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/docker-compose.yml b/docker-compose.yml index 58cb1c5..b931d2d 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -9,6 +9,9 @@ services: HOST: "0.0.0.0" PORT: "8000" MPLBACKEND: Agg + volumes: + - ./watchlist.yaml:/app/watchlist.yaml:ro + - ./data:/app/data restart: unless-stopped networks: default: diff --git a/requirements.txt b/requirements.txt index a3b06cc..5c012ae 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,6 +4,8 @@ pydantic>=2.9.0 pydantic-settings>=2.6.0 httpx>=0.27.0 pandas>=2.2.0 +numpy>=2.0.0 +pyyaml>=6.0.2 mplfinance>=0.12.10b0 matplotlib>=3.9.0 python-multipart>=0.0.12 diff --git a/watchlist.yaml b/watchlist.yaml new file mode 100644 index 0000000..d7e01e0 --- /dev/null +++ b/watchlist.yaml @@ -0,0 +1,19 @@ +# Binance USDT-M perpetual symbols (no .P suffix). Edit without rebuilding if the +# file is volume-mounted (see docker-compose.yml). +symbols: + - BTCUSDT + - ETHUSDT + - SOLUSDT + - ZECUSDT + +strategies: + ltf: + enabled: true + timeframe: 15m + # 15m forum topic id (integer from the Telegram topic URL) + telegram_thread_id: 51247 + fvg: + enabled: true + timeframe: 6h + # 6h forum topic id — replace with the other thread + telegram_thread_id: 61476