"""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)"} # Period labels per level per timeframe _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] = [] # โ”€โ”€ minimal summary โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ lines.append( f"๐ŸŒŠ {sig.ticker} | {sig.timeframe} | " f"{sig_emoji} {sig_label}" ) lines.append( f"Price: {_fmt_price(sig.current_price)} โฑ close last bar " f"Vol(60d): {sig.vol_ann:.1%}" ) 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: {sig.midband_slope:+.2e} {slope_dir} " f"Sized pos: {sig.sized_position:+.2f}ร—" ) 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("") # โ”€โ”€ per-level breakdown โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ lines.append("๐Ÿ”ฌ MODWT Decomposition (sym8, all levels)") periods = _PERIODS.get(sig.timeframe, [""] * 7) sig_set = {f"D{j}" for j in sig.sig_levels} header = f"{'Lvl':<4} {'Period':<10} {'Dir':<5} {'Slope bar':<7} {'Description'}" lines.append(header) lines.append("" + "โ”€" * 48 + "") 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"{label:<4} {period:<10} {d_emoji} [{bar}] {desc}{is_signal}" ) lines.append("") lines.append( f"โ—€ signal levels ({d_sig_label}) | " f"bar position: left=bear ยท right=bull" ) lines.append("") # โ”€โ”€ consensus note โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ 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("") # โ”€โ”€ warnings โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ if sig.warnings: lines.append("โš ๏ธ Warnings:") for w in sig.warnings: lines.append(f" {w}") lines.append("") lines.append( f"{sig.timestamp.strftime('%Y-%m-%d %H:%M UTC')} | " f"MODWT {d_sig_label} signal | no PyWavelets | not financial advice" ) return "\n".join(lines)