"""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)