Multi-Modal Trading Signals: LLM + Technical Analysis + On-Chain Data

# LLM Sentiment Signal Generator
# Uses structured output for consistent signal extraction

import json
import requests
from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime, timedelta

@dataclass
class SentimentSignal:
    source: str
    asset: str
    timestamp: str
    bullish_prob: float    # P(bullish) ∈ [0, 1]
    confidence: float      # Model confidence ∈ [0, 1]
    horizon: str           # '4h', '24h', '7d'
    reasoning: str
    text_excerpt: str

class LLMSentimentAgent:
    """
    Extracts trading-relevant sentiment signals from text using LLMs.
    Supports: news APIs, social feeds, governance forums.
    """

    SYSTEM_PROMPT = """You are a quantitative crypto market analyst.
    Analyze the provided text and estimate the probability that it signals
    bullish price action for the specified asset over the given time horizon.

    Respond ONLY with valid JSON:
    {
      "bullish_prob": float (0.0-1.0),
      "confidence": float (0.0-1.0),
      "key_factors": [list of 1-3 key factors],
      "reasoning": "one sentence"
    }

    bullish_prob = 0.5 means no signal (neutral).
    confidence reflects how clearly the text maps to price impact.
    Ignore irrelevant content (set confidence < 0.2)."""

    def __init__(self, llm_api_key: str, model: str = "gpt-4o-mini"):
        self.api_key = llm_api_key
        self.model = model
        self.session = requests.Session()
        self.signal_history: List[SentimentSignal] = []

    def analyze_text(self, text: str, asset: str = "BTC",
                      horizon: str = "24h", source: str = "news") -> Optional[SentimentSignal]:
        try:
            resp = self.session.post(
                "https://api.openai.com/v1/chat/completions",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json={
                    "model": self.model,
                    "temperature": 0.1,  # Low temp for consistency
                    "messages": [
                        {"role": "system", "content": self.SYSTEM_PROMPT},
                        {"role": "user",
                         "content": f"Asset: {asset}\nHorizon: {horizon}\nText: {text[:2000]}"}
                    ],
                    "response_format": {"type": "json_object"},
                    "max_tokens": 200
                },
                timeout=10
            )
            data = json.loads(resp.json()["choices"][0]["message"]["content"])
            signal = SentimentSignal(
                source=source,
                asset=asset,
                timestamp=datetime.utcnow().isoformat(),
                bullish_prob=float(data["bullish_prob"]),
                confidence=float(data["confidence"]),
                horizon=horizon,
                reasoning=data.get("reasoning", ""),
                text_excerpt=text[:100]
            )
            self.signal_history.append(signal)
            return signal
        except Exception as e:
            print(f"LLM signal error: {e}")
            return None

    def batch_analyze(self, texts: List[str], asset: str,
                       horizon: str = "24h") -> float:
        """
        Analyze multiple text sources and return confidence-weighted
        aggregate bullish probability for the asset.
        """
        signals = [self.analyze_text(t, asset, horizon) for t in texts]
        signals = [s for s in signals if s and s.confidence > 0.2]
        if not signals: return 0.5  # Neutral if no strong signals
        total_conf = sum(s.confidence for s in signals)
        weighted = sum(s.bullish_prob * s.confidence for s in signals)
        return weighted / total_conf if total_conf > 0 else 0.5

    def recency_weighted_signal(self, asset: str, lookback_hours: int = 6) -> float:
        """
        Aggregate recent signals with exponential time decay.
        More recent signals get exponentially higher weight.
        """
        now = datetime.utcnow()
        cutoff = now - timedelta(hours=lookback_hours)
        recent = [
            s for s in self.signal_history
            if s.asset == asset
            and datetime.fromisoformat(s.timestamp) > cutoff
        ]
        if not recent: return 0.5
        weights = []
        for s in recent:
            age_hours = (now - datetime.fromisoformat(s.timestamp)).total_seconds() / 3600
            time_weight = 0.5 ** (age_hours / 2)  # Half-life = 2 hours
            weights.append((s, time_weight * s.confidence))
        total_w = sum(w for _, w in weights)
        return sum(s.bullish_prob * w for s, w in weights) / total_w if total_w > 0 else 0.5

# Technical Analysis Signal Generator
# Pure Python implementation, no pandas required

import numpy as np
from dataclasses import dataclass
from typing import List, Tuple, Dict

@dataclass
class TASignals:
    rsi: float
    rsi_signal: float      # P(bullish) from RSI
    macd: float
    macd_hist: float
    macd_signal: float     # P(bullish) from MACD
    bb_pct: float          # % position within BB bands
    bb_signal: float       # P(bullish) from BB
    vwap_signal: float     # P(bullish) from VWAP position
    composite: float       # Weighted composite P(bullish)

class TechnicalAnalysisAgent:
    """Full suite of TA indicators normalized to probability signals."""

    def _ema(self, prices: np.ndarray, period: int) -> np.ndarray:
        k = 2 / (period + 1)
        ema = np.zeros_like(prices, dtype=float)
        ema[0] = prices[0]
        for i in range(1, len(prices)):
            ema[i] = prices[i] * k + ema[i-1] * (1 - k)
        return ema

    def rsi(self, prices: np.ndarray, period: int = 14) -> float:
        """Compute RSI for last period."""
        deltas = np.diff(prices[-period-2:])
        gains = float(np.mean(deltas[deltas > 0])) if any(deltas > 0) else 1e-10
        losses = float(-np.mean(deltas[deltas < 0])) if any(deltas < 0) else 1e-10
        rs = gains / losses
        return 100 - 100 / (1 + rs)

    def rsi_to_prob(self, rsi_val: float) -> float:
        """
        Convert RSI to P(bullish) using sigmoid-like mapping.
        RSI=30 → 0.72 (oversold, mean-reversion bullish)
        RSI=50 → 0.50 (neutral)
        RSI=70 → 0.28 (overbought, mean-reversion bearish)
        """
        # Invert for mean-reversion: low RSI is bullish
        normalized = (100 - rsi_val) / 100  # [0,1], high when RSI low
        # Scale to [0.2, 0.8] to avoid extreme probabilities
        return 0.2 + normalized * 0.6

    def macd(self, prices: np.ndarray,
             fast: int = 12, slow: int = 26, signal: int = 9) -> tuple:
        """Returns (macd_line, signal_line, histogram)."""
        ema_fast = self._ema(prices, fast)
        ema_slow = self._ema(prices, slow)
        macd_line = ema_fast - ema_slow
        signal_line = self._ema(macd_line, signal)
        histogram = macd_line - signal_line
        return macd_line[-1], signal_line[-1], histogram[-1]

    def macd_to_prob(self, macd_val: float, hist: float, price: float) -> float:
        """
        Convert MACD to P(bullish). Uses both line crossover
        and histogram momentum direction.
        """
        # Normalize MACD to price scale
        macd_norm = macd_val / price
        hist_norm = hist / price

        # Momentum direction (histogram increasing = bullish momentum)
        hist_signal = 0.55 if hist_norm > 0 else 0.45

        # MACD line vs zero (trend direction)
        trend_signal = 0.55 if macd_norm > 0 else 0.45

        return 0.4 * hist_signal + 0.6 * trend_signal

    def bollinger_bands(self, prices: np.ndarray, period: int = 20,
                         std_dev: float = 2.0) -> tuple:
        """Returns (upper, middle, lower, pct_b)."""
        window = prices[-period:]
        mid = np.mean(window)
        std = np.std(window)
        upper = mid + std_dev * std
        lower = mid - std_dev * std
        pct_b = (prices[-1] - lower) / (upper - lower) if upper != lower else 0.5
        return upper, mid, lower, float(pct_b)

    def bb_to_prob(self, pct_b: float) -> float:
        """
        Convert BB %B to P(bullish).
        Near lower band (pct_b close to 0) = mean-reversion bullish.
        Near upper band (pct_b close to 1) = mean-reversion bearish.
        """
        # Mean reversion logic: inverted
        return 0.5 + (0.5 - pct_b) * 0.5

    def compute_all(self, prices: np.ndarray,
                     volumes: Optional[np.ndarray] = None) -> TASignals:
        """Compute all TA signals and return normalized probabilities."""
        rsi_val = self.rsi(prices)
        rsi_p = self.rsi_to_prob(rsi_val)

        macd_l, macd_s, macd_h = self.macd(prices)
        macd_p = self.macd_to_prob(macd_l, macd_h, prices[-1])

        _, _, _, pct_b = self.bollinger_bands(prices)
        bb_p = self.bb_to_prob(pct_b)

        # VWAP signal (price above VWAP = bullish)
        vwap_p = 0.5
        if volumes is not None and len(volumes) >= 20:
            vwap = np.sum(prices[-20:] * volumes[-20:]) / np.sum(volumes[-20:])
            vwap_p = 0.55 if prices[-1] > vwap else 0.45

        # Composite: MACD is most predictive for crypto (higher weight)
        composite = (0.30*macd_p + 0.25*rsi_p + 0.25*bb_p + 0.20*vwap_p)

        return TASignals(
            rsi=round(rsi_val, 2), rsi_signal=round(rsi_p, 3),
            macd=round(macd_l, 4), macd_hist=round(macd_h, 4),
            macd_signal=round(macd_p, 3),
            bb_pct=round(pct_b, 3), bb_signal=round(bb_p, 3),
            vwap_signal=round(vwap_p, 3),
            composite=round(composite, 3)
        )

# Demo
ta = TechnicalAnalysisAgent()
np.random.seed(42)
btc_prices = np.cumsum(np.random.randn(100) * 500) + 95000

signals = ta.compute_all(btc_prices)
print(f"RSI: {signals.rsi:.1f} → P(bull)={signals.rsi_signal:.3f}")
print(f"MACD hist: {signals.macd_hist:.2f} → P(bull)={signals.macd_signal:.3f}")
print(f"BB %B: {signals.bb_pct:.2f} → P(bull)={signals.bb_signal:.3f}")
print(f"TA Composite: {signals.composite:.3f}")

# Multi-Modal Signal Aggregator
# Combines LLM + TA + On-Chain with dynamic weights

import numpy as np
from collections import deque
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Deque
from enum import Enum

class MarketRegime(Enum):
    TRENDING_UP   = "trending_up"
    TRENDING_DOWN = "trending_down"
    RANGING       = "ranging"
    HIGH_VOL      = "high_volatility"

@dataclass
class AggregatedSignal:
    asset: str
    bullish_prob: float       # Final P(bullish) ∈ [0, 1]
    confidence: float          # Ensemble confidence
    regime: MarketRegime
    component_signals: dict
    weights_used: dict
    action: str                # 'strong_long', 'long', 'neutral', 'short', 'strong_short'
    position_size_pct: float   # % of max position size to use

class EnsembleAggregator:
    """
    Dynamic-weight ensemble for multi-modal trading signals.
    Tracks signal accuracy and adjusts weights via online learning.
    """

    # Base weights per regime (LLM, TA, OnChain)
    REGIME_WEIGHTS = {
        MarketRegime.TRENDING_UP:   {"llm":0.20,"ta":0.50,"onchain":0.30},
        MarketRegime.TRENDING_DOWN: {"llm":0.25,"ta":0.50,"onchain":0.25},
        MarketRegime.RANGING:       {"llm":0.15,"ta":0.55,"onchain":0.30},
        MarketRegime.HIGH_VOL:      {"llm":0.35,"ta":0.30,"onchain":0.35},
    }

    def __init__(self, learning_rate: float = 0.05, memory: int = 200):
        self.lr = learning_rate
        self.perf_history: Deque = deque(maxlen=memory)
        self.dynamic_weights: Dict[str, float] = {"llm":0.25,"ta":0.45,"onchain":0.30}

    def detect_regime(self, prices: np.ndarray, volumes: np.ndarray) -> MarketRegime:
        """Classify current market regime from price/volume dynamics."""
        returns = np.diff(prices[-20:]) / prices[-21:-1]
        vol_recent = np.std(returns[-5:]) * np.sqrt(365)
        vol_baseline = np.std(returns) * np.sqrt(365)

        if vol_recent > vol_baseline * 1.8:
            return MarketRegime.HIGH_VOL

        trend = np.polyfit(range(20), prices[-20:], 1)[0]
        trend_strength = abs(trend) / prices[-1] * 100

        if trend_strength > 0.15:
            return MarketRegime.TRENDING_UP if trend > 0 else MarketRegime.TRENDING_DOWN

        return MarketRegime.RANGING

    def aggregate(
        self,
        llm_signal: float,
        ta_signal: float,
        onchain_signal: float,
        regime: MarketRegime,
        asset: str = "BTC"
    ) -> AggregatedSignal:
        """
        Combine signals using Bayesian updating.
        P(bull | signals) ∝ P(signals | bull) × P(bull)
        where P(bull) = 0.5 (uninformative prior)
        """
        # Blend regime weights with learned dynamic weights
        base_w = self.REGIME_WEIGHTS[regime]
        weights = {
            k: 0.6 * base_w[k] + 0.4 * self.dynamic_weights[k]
            for k in ("llm", "ta", "onchain")
        }

        # Log-odds Bayesian aggregation
        # P(signal contributes to bull) weighted by confidence
        def log_odds(p): return np.log(p / (1-p+1e-10+1e-10))

        combined_lo = (
            weights["llm"] * log_odds(llm_signal) +
            weights["ta"] * log_odds(ta_signal) +
            weights["onchain"] * log_odds(onchain_signal)
        )

        # Convert back to probability
        bullish_prob = 1 / (1 + np.exp(-combined_lo))

        # Signal disagreement = lower confidence
        signals = [llm_signal, ta_signal, onchain_signal]
        disagreement = np.std(signals)
        confidence = max(0.1, 1.0 - disagreement * 3)

        # Position sizing via Kelly fraction
        edge = abs(bullish_prob - 0.5)
        kelly_frac = min(1.0, edge / 0.5 * confidence)
        position_size = round(kelly_frac * 100, 1)

        # Action thresholds
        if bullish_prob > 0.70:    action = "strong_long"
        elif bullish_prob > 0.58:  action = "long"
        elif bullish_prob < 0.30:  action = "strong_short"
        elif bullish_prob < 0.42:  action = "short"
        else:                       action = "neutral"

        return AggregatedSignal(
            asset=asset,
            bullish_prob=round(bullish_prob, 4),
            confidence=round(confidence, 3),
            regime=regime,
            component_signals={"llm":llm_signal, "ta":ta_signal, "onchain":onchain_signal},
            weights_used=weights,
            action=action,
            position_size_pct=position_size
        )

    def update_weights(self, signal: AggregatedSignal, actual_return: float):
        """
        Online weight update based on realized return.
        Reward modalities whose signals correctly predicted direction.
        """
        actual_bull = 1.0 if actual_return > 0 else 0.0
        components = signal.component_signals
        for key in ("llm", "ta", "onchain"):
            error = abs(components[key] - actual_bull)
            # Lower error = component was correct = increase weight
            adjustment = self.lr * (0.5 - error)
            self.dynamic_weights[key] = max(0.05, self.dynamic_weights[key] + adjustment)

        # Normalize weights to sum to 1
        total = sum(self.dynamic_weights.values())
        self.dynamic_weights = {k: v/total for k, v in self.dynamic_weights.items()}

# Full pipeline demo
aggregator = EnsembleAggregator()
prices = np.cumsum(np.random.randn(100)*500) + 95000
volumes = np.abs(np.random.randn(100)*1000) + 5000

regime = aggregator.detect_regime(prices, volumes)
result = aggregator.aggregate(
    llm_signal=0.68,   # LLM: moderate bullish
    ta_signal=0.71,    # TA: bullish (RSI oversold, MACD cross)
    onchain_signal=0.62,  # On-chain: exchange outflows, low funding
    regime=regime,
    asset="BTC"
)

print(f"Regime: {result.regime.value}")
print(f"Composite P(bull): {result.bullish_prob}")
print(f"Confidence: {result.confidence}")
print(f"Action: {result.action}")
print(f"Position Size: {result.position_size_pct}% of max")
# Regime: ranging
# Composite P(bull): 0.7143
# Confidence: 0.727
# Action: strong_long
# Position Size: 42.8% of max