Spaces:

ResearchEngineering
/

financial_news_bot

Running

financial_news_bot / src /transformations /wavelets_lite /interpreter.py

Dmitry Beresnev

fix valuation module, etc

e4ed34c 4 months ago

10.6 kB

"""
Frequency component interpreter for wavelets_lite.
Takes a LiteSignal (or raw MODWT arrays) and maps the zone alignment into one
of 12 named market patterns, a conviction score, and an action bias.
Zones:
    Zone 1 (noise):  D1, D2, D3
    Zone 2 (signal): sig_levels  (default D4, D5)
    Zone 3 (trend):  D6, A6
Public API:
    interpret_signal(sig: LiteSignal) -> FrequencyInterpretation
    interpret_signal_from_arrays(details, approx, timeframe, sig_levels) -> FrequencyInterpretation
    format_interpretation(interp: FrequencyInterpretation) -> str
"""
from __future__ import annotations

from dataclasses import dataclass

import numpy as np

from .analyzer import LiteSignal
from .core import linear_slope, midband as _midband, safe_slope_window


# ── Pattern registry ───────────────────────────────────────────────────────────

_PATTERN_DESC: dict[str, str] = {
    "FULL_BULL":                "All time scales trending up — strongest bull alignment",
    "FULL_BEAR":                "All time scales trending down — strongest bear alignment",
    "BULL_WITH_NOISE_HEADWIND": "Pullback within a multi-scale uptrend — best long entry",
    "BEAR_WITH_NOISE_TAILWIND": "Bounce within a multi-scale downtrend — best short entry",
    "COUNTER_TREND_RALLY":      "Intermediate rally against a structural downtrend — caution on longs",
    "PULLBACK_IN_UPTREND":      "Intermediate correction within a structural uptrend — dip-buy candidate",
    "ACCELERATION":             "Momentum building: noise and signal aligned, structural trend turning up",
    "TREND_EXHAUSTION":         "Mid-band energy fading while structural uptrend still holds",
    "BEAR_EXHAUSTION":          "Downtrend losing mid-band energy — potential base forming",
    "TRANSITION":               "Mid-band stalled, short-term bounce against structural downtrend",
    "NOISE_REGIME":             "No mid-band or structural trend — sideways consolidation",
    "STRUCTURAL_DIVERGENCE":    "Intermediate trend up but semi-annual component weakening",
    "UNKNOWN":                  "Zone combination does not match any known pattern",
}

_PATTERN_ACTION: dict[str, str] = {
    "FULL_BULL":                "FULL_SIZE",
    "FULL_BEAR":                "FULL_SIZE",
    "BULL_WITH_NOISE_HEADWIND": "FULL_SIZE",
    "BEAR_WITH_NOISE_TAILWIND": "FULL_SIZE",
    "COUNTER_TREND_RALLY":      "HALF_SIZE",
    "PULLBACK_IN_UPTREND":      "HALF_SIZE",
    "ACCELERATION":             "FULL_SIZE",
    "TREND_EXHAUSTION":         "FADE",
    "BEAR_EXHAUSTION":          "FADE",
    "TRANSITION":               "PASS",
    "NOISE_REGIME":             "PASS",
    "STRUCTURAL_DIVERGENCE":    "HALF_SIZE",
    "UNKNOWN":                  "PASS",
}

_ZONE1_LABELS = ["D1", "D2", "D3"]
_ZONE3_LABELS = ["D6", "A6"]

_ZONE_WEIGHTS = (0.15, 0.35, 0.50)  # Z1, Z2, Z3


# ── Dataclass ──────────────────────────────────────────────────────────────────

@dataclass
class FrequencyInterpretation:
    """Result of a frequency-zone pattern analysis.
    Attributes:
        pattern:     Named market pattern (one of 12 + UNKNOWN).
        conviction:  0.0–1.0 weighted zone alignment score.
        action_bias: FULL_SIZE / HALF_SIZE / PASS / FADE.
        direction:   UP / DOWN / FLAT — derived from Zone 2 (mid-band).
        description: One-line human-readable description of the pattern.
        zone1_vote:  Majority direction of D1, D2, D3 (noise zone).
        zone2_vote:  Direction of the mid-band signal levels (signal zone).
        zone3_vote:  Majority direction of D6, A6 (trend zone).
    """
    pattern:     str
    conviction:  float
    action_bias: str
    direction:   str
    description: str
    zone1_vote:  str
    zone2_vote:  str
    zone3_vote:  str


# ── Internal helpers ───────────────────────────────────────────────────────────

def _zone_vote(signals: dict[str, float], labels: list[str]) -> str:
    """Majority vote of level_signals over the given labels."""
    values = [signals[lb] for lb in labels if lb in signals]
    if not values:
        return "FLAT"
    up   = sum(1 for v in values if v > 0)
    down = sum(1 for v in values if v < 0)
    flat = len(values) - up - down
    if up > down and up > flat:
        return "UP"
    if down > up and down > flat:
        return "DOWN"
    return "FLAT"


def _sig_vote(raw_signal: float) -> str:
    if raw_signal > 0:
        return "UP"
    if raw_signal < 0:
        return "DOWN"
    return "FLAT"


def _detect_pattern(z1: str, z2: str, z3: str) -> str:
    """Map (zone1, zone2, zone3) votes to a named pattern."""
    if z2 == "FLAT":
        if z3 == "FLAT":
            return "NOISE_REGIME"
        if z3 == "UP":
            return "TREND_EXHAUSTION"
        # z3 == "DOWN"
        return "TRANSITION" if z1 == "UP" else "BEAR_EXHAUSTION"

    if z2 == "UP":
        if z3 == "UP":
            return "FULL_BULL" if z1 == "UP" else "BULL_WITH_NOISE_HEADWIND"
        if z3 == "DOWN":
            return "COUNTER_TREND_RALLY"
        # z3 == "FLAT"
        return "ACCELERATION" if z1 == "UP" else "STRUCTURAL_DIVERGENCE"

    # z2 == "DOWN"
    if z3 == "DOWN":
        return "FULL_BEAR" if z1 == "DOWN" else "BEAR_WITH_NOISE_TAILWIND"
    if z3 == "UP":
        return "PULLBACK_IN_UPTREND"
    # z3 == "FLAT"
    return "BEAR_EXHAUSTION"


def _agreement(zone_vote: str, direction: str) -> float:
    """Zone agreement score: 1.0 (agrees) / 0.5 (flat) / 0.0 (opposes)."""
    if direction == "FLAT" or zone_vote == "FLAT":
        return 0.5
    return 1.0 if zone_vote == direction else 0.0


def _compute_conviction(z1: str, z2: str, z3: str) -> float:
    """Weighted zone alignment — Zone 3 (0.50) > Zone 2 (0.35) > Zone 1 (0.15)."""
    if z2 == "FLAT":
        return 0.0
    direction = "UP" if z2 == "UP" else "DOWN"
    return round(
        _agreement(z1, direction) * _ZONE_WEIGHTS[0]
        + 1.0                     * _ZONE_WEIGHTS[1]  # Z2 always agrees with itself
        + _agreement(z3, direction) * _ZONE_WEIGHTS[2],
        3,
    )


def _build(z1: str, z2: str, z3: str) -> FrequencyInterpretation:
    pattern = _detect_pattern(z1, z2, z3)
    return FrequencyInterpretation(
        pattern     = pattern,
        conviction  = _compute_conviction(z1, z2, z3),
        action_bias = _PATTERN_ACTION[pattern],
        direction   = z2,
        description = _PATTERN_DESC[pattern],
        zone1_vote  = z1,
        zone2_vote  = z2,
        zone3_vote  = z3,
    )


# ── Public API ─────────────────────────────────────────────────────────────────

def interpret_signal(sig: LiteSignal) -> FrequencyInterpretation:
    """Interpret a LiteSignal's frequency zones into a named market pattern.
    Uses precomputed level_signals from the LiteSignal — no re-computation.
    Args:
        sig: Output of WaveletLiteAnalyzer.analyze() or _analyze_sync().
    Returns:
        FrequencyInterpretation with pattern, conviction, action_bias, votes.
    """
    z1 = _zone_vote(sig.level_signals, _ZONE1_LABELS)
    z2 = _sig_vote(sig.raw_signal)
    z3 = _zone_vote(sig.level_signals, _ZONE3_LABELS)
    return _build(z1, z2, z3)


def interpret_signal_from_arrays(
    details: dict[int, np.ndarray],
    approx: np.ndarray,
    sig_levels: list[int],
    slope_window: int = 40,
) -> FrequencyInterpretation:
    """Interpret raw MODWT arrays into a named market pattern.
    Use this when you have atrous_swt() output but no LiteSignal.
    Args:
        details:      {j: D_j array} dict from atrous_swt().
        approx:       Final approximation array A_N from atrous_swt().
        sig_levels:   Detail levels that form the mid-band, e.g. [4, 5].
        slope_window: Bars for OLS slope estimation (default 40).
    Returns:
        FrequencyInterpretation with pattern, conviction, action_bias, votes.
    """
    n             = len(approx)
    decomp_levels = max(details.keys()) if details else 6

    level_signals: dict[str, float] = {}
    for j in sorted(details.keys()):
        sw = safe_slope_window(j, n, slope_window)
        level_signals[f"D{j}"] = float(np.sign(linear_slope(details[j], sw)))

    sw_a = safe_slope_window(decomp_levels, n, slope_window)
    level_signals["A6"] = float(np.sign(linear_slope(approx, sw_a)))

    mb       = _midband(details, sig_levels)
    sw_mb    = safe_slope_window(max(sig_levels), n, slope_window)
    raw      = float(np.sign(linear_slope(mb, sw_mb)))

    z1 = _zone_vote(level_signals, _ZONE1_LABELS)
    z2 = _sig_vote(raw)
    z3 = _zone_vote(level_signals, _ZONE3_LABELS)
    return _build(z1, z2, z3)


# ── Formatter ──────────────────────────────────────────────────────────────────

_ACTION_EMOJI = {
    "FULL_SIZE": "🟢",
    "HALF_SIZE": "🟡",
    "PASS":      "⚫",
    "FADE":      "🔴",
}
_DIR_ARROW  = {"UP": "▲", "DOWN": "▼", "FLAT": "→"}
_ZONE_EMOJI = {"UP": "🟢", "DOWN": "🔴", "FLAT": "⚫"}


def format_interpretation(interp: FrequencyInterpretation) -> str:
    """Render a FrequencyInterpretation as a Telegram HTML string."""
    action_emoji = _ACTION_EMOJI.get(interp.action_bias, "❓")
    dir_arrow    = _DIR_ARROW.get(interp.direction, "→")

    z1e = _ZONE_EMOJI.get(interp.zone1_vote, "⚫")
    z2e = _ZONE_EMOJI.get(interp.zone2_vote, "⚫")
    z3e = _ZONE_EMOJI.get(interp.zone3_vote, "⚫")

    lines: list[str] = [
        f"📊 <b>Pattern:</b> <code>{interp.pattern}</code>",
        f"   {dir_arrow} <b>{interp.direction}</b>  "
        f"conviction <b>{interp.conviction:.0%}</b>  "
        f"{action_emoji} <b>{interp.action_bias}</b>",
        f"   <i>{interp.description}</i>",
        "",
        "<code>Zone 1 (noise)   D1–D3 </code>" + f"{z1e} {interp.zone1_vote}",
        "<code>Zone 2 (signal)  D4–D5 </code>" + f"{z2e} {interp.zone2_vote}",
        "<code>Zone 3 (trend)   D6+A6 </code>" + f"{z3e} {interp.zone3_vote}",
    ]
    return "\n".join(lines)

	"""
	Frequency component interpreter for wavelets_lite.

	Takes a LiteSignal (or raw MODWT arrays) and maps the zone alignment into one
	of 12 named market patterns, a conviction score, and an action bias.

	Zones:
	Zone 1 (noise): D1, D2, D3
	Zone 2 (signal): sig_levels (default D4, D5)
	Zone 3 (trend): D6, A6

	Public API:
	interpret_signal(sig: LiteSignal) -> FrequencyInterpretation
	interpret_signal_from_arrays(details, approx, timeframe, sig_levels) -> FrequencyInterpretation
	format_interpretation(interp: FrequencyInterpretation) -> str
	"""
	from __future__ import annotations

	from dataclasses import dataclass

	import numpy as np

	from .analyzer import LiteSignal
	from .core import linear_slope, midband as _midband, safe_slope_window


	# ── Pattern registry ───────────────────────────────────────────────────────────

	_PATTERN_DESC: dict[str, str] = {
	"FULL_BULL": "All time scales trending up — strongest bull alignment",
	"FULL_BEAR": "All time scales trending down — strongest bear alignment",
	"BULL_WITH_NOISE_HEADWIND": "Pullback within a multi-scale uptrend — best long entry",
	"BEAR_WITH_NOISE_TAILWIND": "Bounce within a multi-scale downtrend — best short entry",
	"COUNTER_TREND_RALLY": "Intermediate rally against a structural downtrend — caution on longs",
	"PULLBACK_IN_UPTREND": "Intermediate correction within a structural uptrend — dip-buy candidate",
	"ACCELERATION": "Momentum building: noise and signal aligned, structural trend turning up",
	"TREND_EXHAUSTION": "Mid-band energy fading while structural uptrend still holds",
	"BEAR_EXHAUSTION": "Downtrend losing mid-band energy — potential base forming",
	"TRANSITION": "Mid-band stalled, short-term bounce against structural downtrend",
	"NOISE_REGIME": "No mid-band or structural trend — sideways consolidation",
	"STRUCTURAL_DIVERGENCE": "Intermediate trend up but semi-annual component weakening",
	"UNKNOWN": "Zone combination does not match any known pattern",
	}

	_PATTERN_ACTION: dict[str, str] = {
	"FULL_BULL": "FULL_SIZE",
	"FULL_BEAR": "FULL_SIZE",
	"BULL_WITH_NOISE_HEADWIND": "FULL_SIZE",
	"BEAR_WITH_NOISE_TAILWIND": "FULL_SIZE",
	"COUNTER_TREND_RALLY": "HALF_SIZE",
	"PULLBACK_IN_UPTREND": "HALF_SIZE",
	"ACCELERATION": "FULL_SIZE",
	"TREND_EXHAUSTION": "FADE",
	"BEAR_EXHAUSTION": "FADE",
	"TRANSITION": "PASS",
	"NOISE_REGIME": "PASS",
	"STRUCTURAL_DIVERGENCE": "HALF_SIZE",
	"UNKNOWN": "PASS",
	}

	_ZONE1_LABELS = ["D1", "D2", "D3"]
	_ZONE3_LABELS = ["D6", "A6"]

	_ZONE_WEIGHTS = (0.15, 0.35, 0.50) # Z1, Z2, Z3


	# ── Dataclass ──────────────────────────────────────────────────────────────────

	@dataclass
	class FrequencyInterpretation:
	"""Result of a frequency-zone pattern analysis.

	Attributes:
	pattern: Named market pattern (one of 12 + UNKNOWN).
	conviction: 0.0–1.0 weighted zone alignment score.
	action_bias: FULL_SIZE / HALF_SIZE / PASS / FADE.
	direction: UP / DOWN / FLAT — derived from Zone 2 (mid-band).
	description: One-line human-readable description of the pattern.
	zone1_vote: Majority direction of D1, D2, D3 (noise zone).
	zone2_vote: Direction of the mid-band signal levels (signal zone).
	zone3_vote: Majority direction of D6, A6 (trend zone).
	"""
	pattern: str
	conviction: float
	action_bias: str
	direction: str
	description: str
	zone1_vote: str
	zone2_vote: str
	zone3_vote: str


	# ── Internal helpers ───────────────────────────────────────────────────────────

	def _zone_vote(signals: dict[str, float], labels: list[str]) -> str:
	"""Majority vote of level_signals over the given labels."""
	values = [signals[lb] for lb in labels if lb in signals]
	if not values:
	return "FLAT"
	up = sum(1 for v in values if v > 0)
	down = sum(1 for v in values if v < 0)
	flat = len(values) - up - down
	if up > down and up > flat:
	return "UP"
	if down > up and down > flat:
	return "DOWN"
	return "FLAT"


	def _sig_vote(raw_signal: float) -> str:
	if raw_signal > 0:
	return "UP"
	if raw_signal < 0:
	return "DOWN"
	return "FLAT"


	def _detect_pattern(z1: str, z2: str, z3: str) -> str:
	"""Map (zone1, zone2, zone3) votes to a named pattern."""
	if z2 == "FLAT":
	if z3 == "FLAT":
	return "NOISE_REGIME"
	if z3 == "UP":
	return "TREND_EXHAUSTION"
	# z3 == "DOWN"
	return "TRANSITION" if z1 == "UP" else "BEAR_EXHAUSTION"

	if z2 == "UP":
	if z3 == "UP":
	return "FULL_BULL" if z1 == "UP" else "BULL_WITH_NOISE_HEADWIND"
	if z3 == "DOWN":
	return "COUNTER_TREND_RALLY"
	# z3 == "FLAT"
	return "ACCELERATION" if z1 == "UP" else "STRUCTURAL_DIVERGENCE"

	# z2 == "DOWN"
	if z3 == "DOWN":
	return "FULL_BEAR" if z1 == "DOWN" else "BEAR_WITH_NOISE_TAILWIND"
	if z3 == "UP":
	return "PULLBACK_IN_UPTREND"
	# z3 == "FLAT"
	return "BEAR_EXHAUSTION"


	def _agreement(zone_vote: str, direction: str) -> float:
	"""Zone agreement score: 1.0 (agrees) / 0.5 (flat) / 0.0 (opposes)."""
	if direction == "FLAT" or zone_vote == "FLAT":
	return 0.5
	return 1.0 if zone_vote == direction else 0.0


	def _compute_conviction(z1: str, z2: str, z3: str) -> float:
	"""Weighted zone alignment — Zone 3 (0.50) > Zone 2 (0.35) > Zone 1 (0.15)."""
	if z2 == "FLAT":
	return 0.0
	direction = "UP" if z2 == "UP" else "DOWN"
	return round(
	_agreement(z1, direction) * _ZONE_WEIGHTS[0]
	+ 1.0 * _ZONE_WEIGHTS[1] # Z2 always agrees with itself
	+ _agreement(z3, direction) * _ZONE_WEIGHTS[2],
	3,
	)


	def _build(z1: str, z2: str, z3: str) -> FrequencyInterpretation:
	pattern = _detect_pattern(z1, z2, z3)
	return FrequencyInterpretation(
	pattern = pattern,
	conviction = _compute_conviction(z1, z2, z3),
	action_bias = _PATTERN_ACTION[pattern],
	direction = z2,
	description = _PATTERN_DESC[pattern],
	zone1_vote = z1,
	zone2_vote = z2,
	zone3_vote = z3,
	)


	# ── Public API ─────────────────────────────────────────────────────────────────

	def interpret_signal(sig: LiteSignal) -> FrequencyInterpretation:
	"""Interpret a LiteSignal's frequency zones into a named market pattern.

	Uses precomputed level_signals from the LiteSignal — no re-computation.

	Args:
	sig: Output of WaveletLiteAnalyzer.analyze() or _analyze_sync().

	Returns:
	FrequencyInterpretation with pattern, conviction, action_bias, votes.
	"""
	z1 = _zone_vote(sig.level_signals, _ZONE1_LABELS)
	z2 = _sig_vote(sig.raw_signal)
	z3 = _zone_vote(sig.level_signals, _ZONE3_LABELS)
	return _build(z1, z2, z3)


	def interpret_signal_from_arrays(
	details: dict[int, np.ndarray],
	approx: np.ndarray,
	sig_levels: list[int],
	slope_window: int = 40,
	) -> FrequencyInterpretation:
	"""Interpret raw MODWT arrays into a named market pattern.

	Use this when you have atrous_swt() output but no LiteSignal.

	Args:
	details: {j: D_j array} dict from atrous_swt().
	approx: Final approximation array A_N from atrous_swt().
	sig_levels: Detail levels that form the mid-band, e.g. [4, 5].
	slope_window: Bars for OLS slope estimation (default 40).

	Returns:
	FrequencyInterpretation with pattern, conviction, action_bias, votes.
	"""
	n = len(approx)
	decomp_levels = max(details.keys()) if details else 6

	level_signals: dict[str, float] = {}
	for j in sorted(details.keys()):
	sw = safe_slope_window(j, n, slope_window)
	level_signals[f"D{j}"] = float(np.sign(linear_slope(details[j], sw)))

	sw_a = safe_slope_window(decomp_levels, n, slope_window)
	level_signals["A6"] = float(np.sign(linear_slope(approx, sw_a)))

	mb = _midband(details, sig_levels)
	sw_mb = safe_slope_window(max(sig_levels), n, slope_window)
	raw = float(np.sign(linear_slope(mb, sw_mb)))

	z1 = _zone_vote(level_signals, _ZONE1_LABELS)
	z2 = _sig_vote(raw)
	z3 = _zone_vote(level_signals, _ZONE3_LABELS)
	return _build(z1, z2, z3)


	# ── Formatter ──────────────────────────────────────────────────────────────────

	_ACTION_EMOJI = {
	"FULL_SIZE": "🟢",
	"HALF_SIZE": "🟡",
	"PASS": "⚫",
	"FADE": "🔴",
	}
	_DIR_ARROW = {"UP": "▲", "DOWN": "▼", "FLAT": "→"}
	_ZONE_EMOJI = {"UP": "🟢", "DOWN": "🔴", "FLAT": "⚫"}


	def format_interpretation(interp: FrequencyInterpretation) -> str:
	"""Render a FrequencyInterpretation as a Telegram HTML string."""
	action_emoji = _ACTION_EMOJI.get(interp.action_bias, "❓")
	dir_arrow = _DIR_ARROW.get(interp.direction, "→")

	z1e = _ZONE_EMOJI.get(interp.zone1_vote, "⚫")
	z2e = _ZONE_EMOJI.get(interp.zone2_vote, "⚫")
	z3e = _ZONE_EMOJI.get(interp.zone3_vote, "⚫")

	lines: list[str] = [
	f"📊 <b>Pattern:</b> <code>{interp.pattern}</code>",
	f" {dir_arrow} <b>{interp.direction}</b> "
	f"conviction <b>{interp.conviction:.0%}</b> "
	f"{action_emoji} <b>{interp.action_bias}</b>",
	f" <i>{interp.description}</i>",
	"",
	"<code>Zone 1 (noise) D1–D3 </code>" + f"{z1e} {interp.zone1_vote}",
	"<code>Zone 2 (signal) D4–D5 </code>" + f"{z2e} {interp.zone2_vote}",
	"<code>Zone 3 (trend) D6+A6 </code>" + f"{z3e} {interp.zone3_vote}",
	]
	return "\n".join(lines)