| """Telegram HTML formatter for LiteSignal.""" |
| from __future__ import annotations |
| from .analyzer import LiteSignal |
|
|
| _SIG_EMOJI = {1.0: "🟢", -1.0: "🔴", 0.0: "⚫"} |
| _SIG_LABEL = {1.0: "LONG (up-trend)", -1.0: "SHORT (down-trend)", 0.0: "FLAT (no signal)"} |
|
|
| |
| _PERIODS: dict[str, list[str]] = { |
| "1h": ["2–4h", "4–8h", "8–16h", "16–32h", "32–64h", "64–128h", ">128h"], |
| "4h": ["8–16h", "16–32h", "1–3d", "3–5d", "5–11d", "11–21d", ">21d"], |
| "1d": ["2–4d", "4–8d", "8–16d", "16–32d", "32–64d", "64–128d", ">128d"], |
| "1wk": ["2–4wk", "4–8wk", "8–16wk", "4–8mo", "8–16mo", "16–32mo", ">32mo"], |
| } |
|
|
| _LEVEL_LABELS = ["D1", "D2", "D3", "D4", "D5", "D6", "A6"] |
| _SCALE_DESC = [ |
| "microstructure", |
| "short reversals", |
| "news swings", |
| "monthly momentum", |
| "quarterly trend", |
| "macro positioning", |
| "secular drift", |
| ] |
|
|
|
|
| def _fmt_price(p: float) -> str: |
| if p >= 1000: |
| return f"{p:,.2f}" |
| if p >= 10: |
| return f"{p:.2f}" |
| if p >= 1: |
| return f"{p:.4f}" |
| return f"{p:.6f}" |
|
|
|
|
| def _slope_bar(slope: float, scale: float = 5e-5) -> str: |
| """5-char visual bar: left=bear, centre=neutral, right=bull.""" |
| norm = max(-1.0, min(1.0, slope / scale)) |
| filled = int((norm + 1) / 2 * 5) |
| bar = "░" * 5 |
| bar = bar[:filled] + "█" + bar[filled + 1:] |
| return bar[:5] |
|
|
|
|
| def format_lite(sig: LiteSignal) -> str: |
| sig_emoji = _SIG_EMOJI.get(sig.raw_signal, "❓") |
| sig_label = _SIG_LABEL.get(sig.raw_signal, "UNKNOWN") |
| lines: list[str] = [] |
|
|
| |
| lines.append( |
| f"🌊 <b>{sig.ticker}</b> | <code>{sig.timeframe}</code> | " |
| f"{sig_emoji} <b>{sig_label}</b>" |
| ) |
| lines.append( |
| f"Price: <b>{_fmt_price(sig.current_price)}</b> <i>⏱ close last bar</i> " |
| f"Vol(60d): <b>{sig.vol_ann:.1%}</b>" |
| ) |
|
|
| d_sig_label = "+".join(f"D{j}" for j in sorted(sig.sig_levels)) |
| slope_dir = "↑" if sig.midband_slope > 0 else ("↓" if sig.midband_slope < 0 else "→") |
| lines.append( |
| f"Mid-band ({d_sig_label}) slope: <b>{sig.midband_slope:+.2e}</b> {slope_dir} " |
| f"Sized pos: <b>{sig.sized_position:+.2f}×</b>" |
| ) |
| lines.append( |
| f"Lookback: {sig.bars_loaded} bars | " |
| f"{sig.last_bar_time.strftime('%Y-%m-%d') if hasattr(sig.last_bar_time, 'strftime') else sig.last_bar_time}" |
| ) |
| lines.append("") |
|
|
| |
| lines.append("🔬 <b>MODWT Decomposition (sym8, all levels)</b>") |
|
|
| periods = _PERIODS.get(sig.timeframe, [""] * 7) |
| sig_set = {f"D{j}" for j in sig.sig_levels} |
|
|
| header = f"<code>{'Lvl':<4} {'Period':<10} {'Dir':<5} {'Slope bar':<7} {'Description'}</code>" |
| lines.append(header) |
| lines.append("<code>" + "─" * 48 + "</code>") |
|
|
| for i, label in enumerate(_LEVEL_LABELS): |
| direction = sig.level_signals.get(label, 0.0) |
| slope = sig.level_slopes.get(label, 0.0) |
| d_emoji = _SIG_EMOJI.get(direction, "⚫") |
| bar = _slope_bar(slope) |
| period = periods[i] if i < len(periods) else "" |
| desc = _SCALE_DESC[i] if i < len(_SCALE_DESC) else "" |
| is_signal = "◀" if label in sig_set else " " |
| lines.append( |
| f"<code>{label:<4} {period:<10} </code>{d_emoji}<code> [{bar}] {desc}{is_signal}</code>" |
| ) |
|
|
| lines.append("") |
| lines.append( |
| f"<i>◀ signal levels ({d_sig_label}) | " |
| f"bar position: left=bear · right=bull</i>" |
| ) |
| lines.append("") |
|
|
| |
| up = sum(1 for v in sig.level_signals.values() if v > 0) |
| down = sum(1 for v in sig.level_signals.values() if v < 0) |
| total = len(sig.level_signals) |
| lines.append( |
| f"Consensus: 🟢 {up}/{total} up · 🔴 {down}/{total} down · " |
| f"⚫ {total - up - down}/{total} flat" |
| ) |
| lines.append("") |
|
|
| |
| if sig.warnings: |
| lines.append("⚠️ <b>Warnings:</b>") |
| for w in sig.warnings: |
| lines.append(f" <i>{w}</i>") |
| lines.append("") |
|
|
| lines.append( |
| f"<i>{sig.timestamp.strftime('%Y-%m-%d %H:%M UTC')} | " |
| f"MODWT {d_sig_label} signal | no PyWavelets | not financial advice</i>" |
| ) |
| return "\n".join(lines) |
|
|
