| """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"<b>{s.name}</b>") |
| lines.append( |
| f" CAGR: <b>{s.cagr:+.2%}</b> | Vol: {s.volatility:.2%} | " |
| f"Sharpe: <b>{s.sharpe:.3f}</b> Sortino: {s.sortino:.3f}" |
| ) |
| lines.append( |
| f" Max DD: <b>{s.max_drawdown:.2%}</b> | 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"<code>{'Metric':<10} {'MODWT':>10} {'B&H':>8} {'SMA':>8}</code>" |
| divider = "<code>" + "─" * 40 + "</code>" |
| table_rows = [header, divider] |
| for name, v_strat, v_bh, v_sma in rows: |
| table_rows.append(f"<code>{name:<10} {v_strat:>10} {v_bh:>8} {v_sma:>8}</code>") |
|
|
| return "\n".join(table_rows) |
|
|
|
|
| def format_analysis(a: WaveletAnalysis, compact: bool = False) -> str: |
| sig = a.signal |
| lines: list[str] = [] |
|
|
| |
| signal_emoji = _SIGNAL_EMOJI.get(sig.raw_signal, "❓") |
| signal_label = _SIGNAL_LABEL.get(sig.raw_signal, "UNKNOWN") |
|
|
| lines.append( |
| f"🌊 <b>Wavelet Analysis: {a.ticker}</b> | <code>{a.timeframe}</code> | " |
| f"{signal_emoji} <b>{signal_label}</b>" |
| ) |
| lines.append( |
| f"Price: <b>{_fmt_price(sig.current_price)}</b> " |
| f"<i>⏱ close last bar</i> " |
| f"Vol(60d ann): <b>{sig.realized_vol_ann:.1%}</b>" |
| ) |
| 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("") |
|
|
| |
| lines.append("📡 <b>MODWT Mid-Band Signal (D4+D5, 16–64 day cycle)</b>") |
|
|
| slope_direction = "↑" if sig.midband_slope > 0 else ("↓" if sig.midband_slope < 0 else "→") |
| lines.append( |
| f" Mid-band slope: <b>{sig.midband_slope:.6f}</b> {slope_direction} " |
| f"(last safe value: {sig.midband_last:.6f})" |
| ) |
|
|
| pos_desc = ( |
| f"<b>{sig.sized_position:+.2f}×</b> notional" |
| if sig.sized_position != 0 else "<b>FLAT (no position)</b>" |
| ) |
| lines.append(f" Sized position: {pos_desc} (vol-targeted to 10% ann.)") |
|
|
| |
| 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("") |
|
|
| |
| if a.strategy_stats is not None and a.bh_stats is not None and a.sma_stats is not None: |
| lines.append("📊 <b>Walk-Forward Backtest Results</b>") |
| lines.append(f" <i>{a.strategy_stats.years:.1f} years | sym8 wavelet | weekly rebalance | 1bp/side cost</i>") |
| lines.append("") |
|
|
| if not compact: |
| lines.append(_comparison_table(a.strategy_stats, a.bh_stats, a.sma_stats)) |
| lines.append("") |
|
|
| |
| strat = a.strategy_stats |
| bh = a.bh_stats |
| sma = a.sma_stats |
|
|
| lines.append("<b>Key takeaways:</b>") |
|
|
| |
| sharpe_vs_bh = strat.sharpe - bh.sharpe |
| sharpe_vs_sma = strat.sharpe - sma.sharpe |
| lines.append( |
| f" • Sharpe <b>{strat.sharpe:.3f}</b> " |
| 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)" |
| ) |
|
|
| |
| if bh.max_drawdown != 0: |
| dd_reduction = (abs(bh.max_drawdown) - abs(strat.max_drawdown)) / abs(bh.max_drawdown) * 100 |
| dd_str = f"{'↓' if dd_reduction > 0 else '↑'}{abs(dd_reduction):.0f}% reduction" |
| else: |
| dd_str = "B&H no drawdown" |
| lines.append( |
| f" • Max DD <b>{strat.max_drawdown:.2%}</b> " |
| f"(vs B&H {bh.max_drawdown:.2%} — {dd_str})" |
| ) |
|
|
| |
| if strat.correlation_to_spy is not None: |
| lines.append( |
| f" • Corr to SPY: <b>{strat.correlation_to_spy:.2f}</b> " |
| f"<i>(diversification vs pure long)</i>" |
| ) |
|
|
| |
| cagr_diff = strat.cagr - bh.cagr |
| lines.append( |
| f" • CAGR <b>{strat.cagr:+.2%}</b> " |
| 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("<i>⚠ Insufficient history for full backtest — signal snapshot only.</i>") |
| lines.append("") |
|
|
| |
| if not compact: |
| lines.append("🔬 <b>MODWT Scale Reference (sym8, 6 levels)</b>") |
| _sig_set = {f"D{j}" for j in a.sig_levels} |
| scale_table = [ |
| ("D1", "2–4d", "microstructure noise"), |
| ("D2", "4–8d", "short reversals"), |
| ("D3", "8–16d", "news-driven swings"), |
| ("D4", "16–32d", "monthly momentum"), |
| ("D5", "32–64d", "quarterly trend"), |
| ("D6", "64–128d", "macro positioning"), |
| ("A6", ">128d", "secular drift"), |
| ] |
| for lvl, period, desc in scale_table: |
| marker = " ← <b>signal</b>" if lvl in _sig_set else "" |
| lines.append(f" <code>{lvl}</code> {period:<10} {desc}{marker}") |
| lines.append("") |
|
|
| |
| if a.warnings: |
| lines.append("⚠️ <b>Warnings:</b>") |
| for w in a.warnings: |
| lines.append(f" <i>{w}</i>") |
| lines.append("") |
|
|
| d_label = "+".join(f"D{j}" for j in sorted(a.sig_levels)) |
| lines.append( |
| f"<i>Generated {a.timestamp.strftime('%Y-%m-%d %H:%M UTC')} | " |
| f"MODWT mid-band ({d_label}) | " |
| "Not financial advice</i>" |
| ) |
|
|
| return "\n".join(lines) |
|
|
