Download src/transformations/wavelets_lite/formatter.py from ResearchEngineering/financial_news_bot: direct link, hf CLI and curl.
- Browser
- Download file 5.22 kB
-
https://huggingface.co/spaces/ResearchEngineering/financial_news_bot/resolve/main/src/transformations/wavelets_lite/formatter.py
- Command line
-
hf download hf://spaces/ResearchEngineering/financial_news_bot/src/transformations/wavelets_lite/formatter.py
-
curl -L -o formatter.py https://huggingface.co/spaces/ResearchEngineering/financial_news_bot/resolve/main/src/transformations/wavelets_lite/formatter.py
5.22 kB
| """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}" | |
| _TF_SLOPE_SCALE: dict[str, float] = { | |
| "1h": 8e-6, | |
| "4h": 2e-5, | |
| "1d": 5e-5, | |
| "1wk": 2e-4, | |
| } | |
| def _slope_bar(slope: float, timeframe: str = "1d") -> str: | |
| """5-char cursor bar: left=bear, centre=neutral, right=bull.""" | |
| scale = _TF_SLOPE_SCALE.get(timeframe, 5e-5) | |
| norm = max(-1.0, min(1.0, slope / scale)) | |
| pos = int((norm + 1) / 2 * 4) # 0..4 β 5 positions, no overflow | |
| bar = list("βββββ") | |
| bar[pos] = "β" | |
| return "".join(bar) | |
| 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"π <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("") | |
| # ββ per-level breakdown ββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| 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, sig.timeframe) | |
| 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("") | |
| # ββ 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("β οΈ <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) | |