"""Telegram HTML formatter for WaveletAnalysis output."""
from __future__ import annotations
import pandas as pd
from .analyzer import WaveletAnalysis, SignalSnapshot
from .stats import Stats
_SIGNAL_EMOJI = {1.0: "🟢 LONG", -1.0: "🔴 SHORT", 0.0: "⚫ FLAT"}
_SIGNAL_LABEL = {1.0: "UP-TREND", -1.0: "DOWN-TREND", 0.0: "NO SIGNAL"}
def _fmt_pct(v: float) -> str:
return f"{v:+.2f}%"
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 _bar(value: float, min_v: float, max_v: float, width: int = 10) -> str:
"""ASCII progress bar scaled between min_v and max_v."""
rng = max_v - min_v
if rng == 0:
filled = width // 2
else:
filled = int((value - min_v) / rng * width)
filled = max(0, min(filled, width))
return "█" * filled + "░" * (width - filled)
def _stats_block(s: Stats) -> str:
lines = []
lines.append(f"{s.name}")
lines.append(
f" CAGR: {s.cagr:+.2%} | Vol: {s.volatility:.2%} | "
f"Sharpe: {s.sharpe:.3f} Sortino: {s.sortino:.3f}"
)
lines.append(
f" Max DD: {s.max_drawdown:.2%} | Calmar: {s.calmar:.3f}"
)
lines.append(
f" Hit Rate: {s.hit_rate:.1%} | Turnover: {s.annual_turnover:.1f}×/yr | "
f"In Market: {s.time_in_market:.1%}"
)
lines.append(
f" Avg Leverage: {s.avg_gross_leverage:.2f}× | "
f"Corr to SPY: {'—' if s.correlation_to_spy is None else f'{s.correlation_to_spy:.2f}'}"
)
return "\n".join(lines)
def _comparison_table(strat: Stats, bh: Stats, sma: Stats) -> str:
rows = [
("CAGR", f"{strat.cagr:+.2%}", f"{bh.cagr:+.2%}", f"{sma.cagr:+.2%}"),
("Vol", f"{strat.volatility:.2%}", f"{bh.volatility:.2%}", f"{sma.volatility:.2%}"),
("Sharpe", f"{strat.sharpe:.3f}", f"{bh.sharpe:.3f}", f"{sma.sharpe:.3f}"),
("Sortino", f"{strat.sortino:.3f}", f"{bh.sortino:.3f}", f"{sma.sortino:.3f}"),
("Max DD", f"{strat.max_drawdown:.2%}", f"{bh.max_drawdown:.2%}", f"{sma.max_drawdown:.2%}"),
("Calmar", f"{strat.calmar:.3f}", f"{bh.calmar:.3f}", f"{sma.calmar:.3f}"),
("Turnover", f"{strat.annual_turnover:.1f}×", f"{bh.annual_turnover:.1f}×", f"{sma.annual_turnover:.1f}×"),
("In Mkt", f"{strat.time_in_market:.1%}", f"{bh.time_in_market:.1%}", f"{sma.time_in_market:.1%}"),
("Corr SPY",
f"{'—' if strat.correlation_to_spy is None else f'{strat.correlation_to_spy:.2f}'}",
"1.00",
f"{'—' if sma.correlation_to_spy is None else f'{sma.correlation_to_spy:.2f}'}"),
]
header = f"{'Metric':<10} {'MODWT':>10} {'B&H':>8} {'SMA':>8}"
divider = "" + "─" * 40 + ""
table_rows = [header, divider]
for name, v_strat, v_bh, v_sma in rows:
table_rows.append(f"{name:<10} {v_strat:>10} {v_bh:>8} {v_sma:>8}")
return "\n".join(table_rows)
def format_analysis(a: WaveletAnalysis, compact: bool = False) -> str:
sig = a.signal
lines: list[str] = []
# ── Header ────────────────────────────────────────────────────────────────
signal_emoji = _SIGNAL_EMOJI.get(sig.raw_signal, "❓")
signal_label = _SIGNAL_LABEL.get(sig.raw_signal, "UNKNOWN")
lines.append(
f"🌊 Wavelet Analysis: {a.ticker} | {a.timeframe} | "
f"{signal_emoji} {signal_label}"
)
lines.append(
f"Price: {_fmt_price(sig.current_price)} "
f"⏱ close last bar "
f"Vol(60d ann): {sig.realized_vol_ann:.1%}"
)
lines.append(
f"Bars used: {sig.bars_used} | "
f"{sig.last_bar_time.strftime('%Y-%m-%d') if hasattr(sig.last_bar_time, 'strftime') else sig.last_bar_time}"
)
lines.append("")
# ── Signal details ────────────────────────────────────────────────────────
lines.append("📡 MODWT Mid-Band Signal (D4+D5, 16–64 day cycle)")
slope_direction = "↑" if sig.midband_slope > 0 else ("↓" if sig.midband_slope < 0 else "→")
lines.append(
f" Mid-band slope: {sig.midband_slope:.6f} {slope_direction} "
f"(last safe value: {sig.midband_last:.6f})"
)
pos_desc = (
f"{sig.sized_position:+.2f}× notional"
if sig.sized_position != 0 else "FLAT (no position)"
)
lines.append(f" Sized position: {pos_desc} (vol-targeted to 10% ann.)")
# Vol bar
vol_bar = _bar(sig.realized_vol_ann, 0.05, 0.50)
lines.append(f" Realized vol: [{vol_bar}] {sig.realized_vol_ann:.1%}")
lines.append("")
# ── Backtest results ──────────────────────────────────────────────────────
if a.strategy_stats is not None and a.bh_stats is not None and a.sma_stats is not None:
lines.append("📊 Walk-Forward Backtest Results")
lines.append(f" {a.strategy_stats.years:.1f} years | sym8 wavelet | weekly rebalance | 1bp/side cost")
lines.append("")
if not compact:
lines.append(_comparison_table(a.strategy_stats, a.bh_stats, a.sma_stats))
lines.append("")
# Highlight key metrics
strat = a.strategy_stats
bh = a.bh_stats
sma = a.sma_stats
lines.append("Key takeaways:")
# Sharpe vs benchmarks
sharpe_vs_bh = strat.sharpe - bh.sharpe
sharpe_vs_sma = strat.sharpe - sma.sharpe
lines.append(
f" • Sharpe {strat.sharpe:.3f} "
f"({'↑' if sharpe_vs_bh > 0 else '↓'}{abs(sharpe_vs_bh):.3f} vs B&H, "
f"{'↑' if sharpe_vs_sma > 0 else '↓'}{abs(sharpe_vs_sma):.3f} vs SMA)"
)
# Drawdown reduction
dd_reduction = (abs(bh.max_drawdown) - abs(strat.max_drawdown)) / abs(bh.max_drawdown) * 100
lines.append(
f" • Max DD {strat.max_drawdown:.2%} "
f"(vs B&H {bh.max_drawdown:.2%} — "
f"{'↓' if dd_reduction > 0 else '↑'}{abs(dd_reduction):.0f}% reduction)"
)
# Correlation
if strat.correlation_to_spy is not None:
lines.append(
f" • Corr to SPY: {strat.correlation_to_spy:.2f} "
f"(diversification vs pure long)"
)
# CAGR parity
cagr_diff = strat.cagr - bh.cagr
lines.append(
f" • CAGR {strat.cagr:+.2%} "
f"({'≈' if abs(cagr_diff) < 0.005 else ('↑' if cagr_diff > 0 else '↓')}"
f"{abs(cagr_diff):.2%} vs B&H), vol {strat.volatility:.1%} "
f"vs {bh.volatility:.1%} B&H"
)
lines.append("")
elif not a.strategy_stats:
lines.append("⚠ Insufficient history for full backtest — signal snapshot only.")
lines.append("")
# ── Wavelet scale reference ────────────────────────────────────────────────
if not compact:
lines.append("🔬 MODWT Scale Reference (sym8, 6 levels)")
scale_table = [
("D1", "2–4d", "microstructure noise"),
("D2", "4–8d", "short reversals"),
("D3", "8–16d", "news-driven swings"),
("D4", "16–32d", "monthly momentum ← signal"),
("D5", "32–64d", "quarterly trend ← signal"),
("D6", "64–128d", "macro positioning"),
("A6", ">128d", "secular drift"),
]
for lvl, period, desc in scale_table:
lines.append(f" {lvl} {period:<10} {desc}")
lines.append("")
# ── Warnings ──────────────────────────────────────────────────────────────
if a.warnings:
lines.append("⚠️ Warnings:")
for w in a.warnings:
lines.append(f" {w}")
lines.append("")
lines.append(
f"Generated {a.timestamp.strftime('%Y-%m-%d %H:%M UTC')} | "
f"MODWT mid-band (D{min(SIG_LEVELS)}+D{max(SIG_LEVELS)}) | "
"Not financial advice"
)
return "\n".join(lines)
# Import after definition to avoid circular reference at module level
from .backtest import SIG_LEVELS