Celes Renata
ca712ad4a0
- Scheduler: lower stale threshold 240→30 min, batch limit 100→500, TTL 14400→3600 - Prediction snapshot: add 24h market_snapshots time-window fallback - Aggregation: add normalize_impact_scores() z-score normalization - Helm: signal-engine replicas → 0 (idle when dual pipeline disabled) - Quality gate: max_snapshot_age_hours 24→48 - Add backfill script for NULL price_at_prediction snapshots - Add PBT bug condition and preservation tests (14 tests)
1269 lines
44 KiB
Python
1269 lines
44 KiB
Python
"""Recency decay, source credibility weighting, and market context
|
||
integration for aggregation.
|
||
|
||
Provides scoring functions used by the aggregation engine to weight
|
||
document intelligence signals when computing trend summaries.
|
||
|
||
Requirements: 2.1–2.6, 3.1–3.5, 4.2–4.3, 5.1–5.7, 6.1–6.5, 16.4–16.5
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import logging
|
||
import math
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
import asyncpg
|
||
|
||
from services.shared.schemas import MarketContext
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Event type base rates for information gain computation (Req 3.1)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
EVENT_TYPE_BASE_RATES: dict[str, float] = {
|
||
"earnings": 0.25,
|
||
"product_launch": 0.10,
|
||
"regulatory": 0.08,
|
||
"legal": 0.05,
|
||
"m_and_a": 0.03,
|
||
"management_change": 0.06,
|
||
"partnership": 0.12,
|
||
"market_expansion": 0.09,
|
||
"restructuring": 0.04,
|
||
"dividend": 0.15,
|
||
}
|
||
DEFAULT_BASE_RATE = 0.1
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ScoringConfig:
|
||
"""Tunable parameters for signal scoring."""
|
||
|
||
# Recency decay: exponential half-life in hours per window.
|
||
# After one half-life, a document's recency weight drops to 0.5.
|
||
half_life_hours: dict[str, float] = field(default_factory=lambda: {
|
||
"intraday": 2.0,
|
||
"1d": 12.0,
|
||
"7d": 72.0,
|
||
"30d": 240.0,
|
||
"90d": 720.0,
|
||
})
|
||
|
||
# Minimum recency weight — prevents very old docs from being zeroed out
|
||
# entirely so they can still contribute trace-level signal.
|
||
min_recency_weight: float = 0.01
|
||
|
||
# Source credibility bounds — credibility scores outside this range
|
||
# are clamped before weighting.
|
||
credibility_floor: float = 0.1
|
||
credibility_ceiling: float = 1.0
|
||
|
||
# Exponent applied to credibility score. >1 penalises low-credibility
|
||
# sources more aggressively; <1 flattens the curve.
|
||
credibility_exponent: float = 1.0
|
||
|
||
# Novelty bonus: multiplier range applied on top of base weight.
|
||
# A novelty_score of 1.0 gets the full bonus; 0.0 gets none.
|
||
novelty_bonus_max: float = 0.25
|
||
|
||
# Confidence floor — documents below this extraction confidence
|
||
# receive zero weight (they are too unreliable to aggregate).
|
||
confidence_floor: float = 0.2
|
||
|
||
# Market context modulation ---
|
||
# When volatility exceeds this threshold (in price units), recency
|
||
# signals are amplified because fast-moving markets make fresh data
|
||
# more important.
|
||
volatility_recency_boost_threshold: float = 1.0
|
||
volatility_recency_boost_max: float = 0.30 # max extra multiplier
|
||
|
||
# When volume surges above this % change, signals get a small boost
|
||
# because high-volume moves carry more conviction.
|
||
volume_surge_threshold_pct: float = 50.0
|
||
volume_surge_boost: float = 0.15
|
||
|
||
# --- Probabilistic scoring parameters ---
|
||
|
||
# Toggle: when True, use probabilistic formulas (sigmoid gate,
|
||
# adaptive decay, info gain, regime multiplier, source accuracy).
|
||
# When False, preserve exact current heuristic behaviour.
|
||
probabilistic: bool = False
|
||
|
||
# Sigmoid gate parameters — smooth replacement for binary confidence gate.
|
||
# Gate value: σ(k·(x - midpoint)) where k = steepness.
|
||
sigmoid_steepness: float = 5.0
|
||
sigmoid_midpoint: float = 0.5
|
||
|
||
# Information gain parameters — surprise weighting for rare events.
|
||
# r = 1 + λ·(-log₂ P(event_type)), clamped to info_gain_max.
|
||
info_gain_lambda: float = 0.3
|
||
info_gain_max: float = 3.0
|
||
default_base_rate: float = 0.1
|
||
|
||
# Adaptive decay parameters — β scaling factors for event-specific
|
||
# half-life adjustment: τ_i = τ_base · (1+β_impact)·(1+β_surprise)·(1+β_market).
|
||
adaptive_decay_impact_scale: float = 1.0
|
||
adaptive_decay_surprise_scale: float = 1.0
|
||
adaptive_decay_market_scale: float = 0.5
|
||
|
||
# Regime multiplier parameters — replaces market context multiplier.
|
||
# M_regime = 1 + regime_return_weight·|z_r| + regime_volume_weight·|z_v|,
|
||
# clamped to [1.0, regime_multiplier_max].
|
||
regime_return_weight: float = 0.15
|
||
regime_volume_weight: float = 0.10
|
||
regime_multiplier_max: float = 2.5
|
||
|
||
|
||
# Singleton default config
|
||
DEFAULT_CONFIG = ScoringConfig()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Recency decay
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def recency_weight(
|
||
published_at: datetime,
|
||
reference_time: datetime,
|
||
window: str,
|
||
config: ScoringConfig = DEFAULT_CONFIG,
|
||
*,
|
||
half_life_override: float | None = None,
|
||
) -> float:
|
||
"""Compute an exponential recency decay weight for a document.
|
||
|
||
Uses the formula: w = 2^(-age_hours / half_life)
|
||
|
||
Args:
|
||
published_at: When the document was published (tz-aware).
|
||
reference_time: The "now" anchor for the aggregation window (tz-aware).
|
||
window: One of the TrendWindow values (e.g. "7d").
|
||
config: Scoring parameters.
|
||
half_life_override: If provided, use this half-life instead of the
|
||
window-based default (used for adaptive decay).
|
||
|
||
Returns:
|
||
A weight in [config.min_recency_weight, 1.0].
|
||
"""
|
||
# Ensure both are tz-aware; treat naive as UTC.
|
||
if published_at.tzinfo is None:
|
||
published_at = published_at.replace(tzinfo=timezone.utc)
|
||
if reference_time.tzinfo is None:
|
||
reference_time = reference_time.replace(tzinfo=timezone.utc)
|
||
|
||
age_seconds = (reference_time - published_at).total_seconds()
|
||
if age_seconds <= 0:
|
||
return 1.0
|
||
|
||
age_hours = age_seconds / 3600.0
|
||
half_life = half_life_override if half_life_override is not None else config.half_life_hours.get(window, 72.0)
|
||
|
||
weight = math.pow(2.0, -age_hours / half_life)
|
||
return max(weight, config.min_recency_weight)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Source credibility weighting
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def credibility_weight(
|
||
source_credibility: float,
|
||
config: ScoringConfig = DEFAULT_CONFIG,
|
||
) -> float:
|
||
"""Compute a weight from a source's credibility score.
|
||
|
||
The raw credibility (0-1) is clamped to [floor, ceiling] then raised
|
||
to ``credibility_exponent``.
|
||
|
||
Args:
|
||
source_credibility: The credibility score from the source or
|
||
document intelligence record (0-1).
|
||
config: Scoring parameters.
|
||
|
||
Returns:
|
||
A weight in [floor^exp, ceiling^exp].
|
||
"""
|
||
clamped = max(config.credibility_floor, min(source_credibility, config.credibility_ceiling))
|
||
return math.pow(clamped, config.credibility_exponent)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Market context adjustment
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def market_context_multiplier(
|
||
market_ctx: MarketContext | None,
|
||
config: ScoringConfig = DEFAULT_CONFIG,
|
||
) -> float:
|
||
"""Compute a multiplicative adjustment from market context features.
|
||
|
||
Returns a value >= 1.0 that amplifies signal weights when market
|
||
conditions suggest heightened importance (high volatility or volume
|
||
surges). Returns 1.0 when no market context is available.
|
||
"""
|
||
if market_ctx is None or not market_ctx.has_data:
|
||
return 1.0
|
||
|
||
boost = 0.0
|
||
|
||
# Volatility boost — more volatile markets make recent signals more valuable
|
||
if market_ctx.volatility is not None and market_ctx.volatility > config.volatility_recency_boost_threshold:
|
||
excess = market_ctx.volatility - config.volatility_recency_boost_threshold
|
||
# Logarithmic scaling so extreme volatility doesn't blow up the weight
|
||
boost += min(
|
||
math.log1p(excess) * 0.15,
|
||
config.volatility_recency_boost_max,
|
||
)
|
||
|
||
# Volume surge boost
|
||
if market_ctx.volume_change_pct is not None and market_ctx.volume_change_pct > config.volume_surge_threshold_pct:
|
||
boost += config.volume_surge_boost
|
||
|
||
return 1.0 + boost
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Sigmoid confidence gate (Req 2.1–2.6)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def sigmoid_gate(
|
||
x: float,
|
||
steepness: float = 5.0,
|
||
midpoint: float = 0.5,
|
||
) -> float:
|
||
"""Smooth sigmoid confidence gate: σ(k·(x - midpoint)).
|
||
|
||
Replaces the binary 0/1 confidence gate in probabilistic mode.
|
||
Returns a value in (0, 1) — higher confidence produces higher gate.
|
||
|
||
Args:
|
||
x: Extraction confidence value, typically in [0, 1].
|
||
steepness: Steepness parameter k (default 5.0).
|
||
midpoint: Midpoint of the sigmoid transition (default 0.5).
|
||
|
||
Returns:
|
||
Gate value in (0, 1).
|
||
"""
|
||
z = steepness * (x - midpoint)
|
||
# Guard against overflow in exp for very negative z
|
||
if z < -500.0:
|
||
return 0.0
|
||
if z > 500.0:
|
||
return 1.0
|
||
return 1.0 / (1.0 + math.exp(-z))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Information gain surprise weighting (Req 3.1–3.5)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def compute_info_gain(
|
||
event_type: str | None,
|
||
lambda_param: float = 0.3,
|
||
max_gain: float = 3.0,
|
||
default_base_rate: float = 0.1,
|
||
) -> float:
|
||
"""Compute information gain factor for an event type.
|
||
|
||
Formula: r = 1 + λ·(-log₂ P(event_type)), clamped to [1.0, max_gain].
|
||
|
||
Rarer events produce higher surprise weight. Unknown event types
|
||
use the default base rate.
|
||
|
||
Args:
|
||
event_type: Event type string (e.g. "earnings", "m_and_a").
|
||
lambda_param: Scaling parameter λ (default 0.3).
|
||
max_gain: Maximum clamp for the info gain factor (default 3.0).
|
||
default_base_rate: Fallback base rate for unknown event types.
|
||
|
||
Returns:
|
||
Information gain factor r in [1.0, max_gain].
|
||
"""
|
||
if event_type is None:
|
||
return 1.0
|
||
|
||
base_rate = EVENT_TYPE_BASE_RATES.get(event_type, default_base_rate)
|
||
# Guard against log₂(0) — base rates must be > 0
|
||
if base_rate <= 0.0:
|
||
base_rate = default_base_rate
|
||
if base_rate <= 0.0:
|
||
return 1.0
|
||
|
||
surprise = -math.log2(base_rate)
|
||
r = 1.0 + lambda_param * surprise
|
||
return min(max(r, 1.0), max_gain)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Adaptive recency decay (Req 5.1–5.7)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def compute_adaptive_half_life(
|
||
base_half_life: float,
|
||
impact_score: float,
|
||
info_gain_factor: float,
|
||
market_multiplier: float,
|
||
config: ScoringConfig,
|
||
) -> float:
|
||
"""Compute adaptive half-life for event-specific recency decay.
|
||
|
||
Formula: τ_i = τ_base · (1 + β_impact) · (1 + β_surprise) · (1 + β_market)
|
||
|
||
The adaptive half-life is always >= base_half_life (decay is never faster).
|
||
|
||
Args:
|
||
base_half_life: Fixed half-life for the window (hours).
|
||
impact_score: Signal impact score in [0, 1].
|
||
info_gain_factor: Information gain factor r in [1.0, 3.0].
|
||
market_multiplier: Market context/regime multiplier in [1.0, ~2.5].
|
||
config: Scoring config with adaptive decay scale parameters.
|
||
|
||
Returns:
|
||
Adaptive half-life in hours, >= base_half_life.
|
||
"""
|
||
# β_impact: impact_score scaled linearly 0→0, 1→adaptive_decay_impact_scale
|
||
beta_impact = impact_score * config.adaptive_decay_impact_scale
|
||
|
||
# β_surprise: info_gain_factor scaled linearly r=1→0, r=3→adaptive_decay_surprise_scale
|
||
beta_surprise = ((info_gain_factor - 1.0) / 2.0) * config.adaptive_decay_surprise_scale
|
||
|
||
# β_market: market_multiplier scaled linearly 1.0→0, 1.45→adaptive_decay_market_scale
|
||
if market_multiplier > 1.0:
|
||
beta_market = ((market_multiplier - 1.0) / 0.45) * config.adaptive_decay_market_scale
|
||
else:
|
||
beta_market = 0.0
|
||
|
||
tau = base_half_life * (1.0 + beta_impact) * (1.0 + beta_surprise) * (1.0 + beta_market)
|
||
# Ensure adaptive half-life is never less than base (Property 5)
|
||
return max(tau, base_half_life)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Regime multiplier (Req 6.1–6.5)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def compute_regime_multiplier(
|
||
returns: list[float] | None,
|
||
volumes: list[float] | None,
|
||
config: ScoringConfig = DEFAULT_CONFIG,
|
||
) -> float:
|
||
"""Compute regime-aware multiplier from return and volume z-scores.
|
||
|
||
Formula: M_regime = 1 + 0.15·|z_r| + 0.10·|z_v|, clamped to [1.0, max].
|
||
|
||
Args:
|
||
returns: List of recent daily returns (at least 20 values for z-score).
|
||
volumes: List of recent daily volumes (at least 20 values for z-score).
|
||
config: Scoring config with regime multiplier parameters.
|
||
|
||
Returns:
|
||
Regime multiplier in [1.0, config.regime_multiplier_max].
|
||
"""
|
||
if not returns or len(returns) < 2:
|
||
return 1.0
|
||
|
||
# Filter out NaN values from returns
|
||
clean_returns = [r for r in returns if not math.isnan(r)]
|
||
if len(clean_returns) < 2:
|
||
return 1.0
|
||
|
||
# Return z-score: z_r = (r_t - μ_20) / σ_20
|
||
r_window = clean_returns[-20:] if len(clean_returns) >= 20 else clean_returns
|
||
r_t = clean_returns[-1]
|
||
mu_r = sum(r_window) / len(r_window)
|
||
var_r = sum((x - mu_r) ** 2 for x in r_window) / len(r_window)
|
||
sigma_r = math.sqrt(var_r)
|
||
|
||
z_r = 0.0
|
||
if sigma_r > 0.0:
|
||
z_r = (r_t - mu_r) / sigma_r
|
||
|
||
# Volume z-score: z_v = (log(V_t) - μ_V) / σ_V
|
||
z_v = 0.0
|
||
if volumes and len(volumes) >= 2:
|
||
clean_volumes = [v for v in volumes if not math.isnan(v)]
|
||
if len(clean_volumes) >= 2:
|
||
v_window = clean_volumes[-20:] if len(clean_volumes) >= 20 else clean_volumes
|
||
# Use log-volumes, guard against zero/negative volumes
|
||
log_vols = [math.log(max(v, 1.0)) for v in v_window]
|
||
log_v_t = math.log(max(clean_volumes[-1], 1.0))
|
||
mu_v = sum(log_vols) / len(log_vols)
|
||
var_v = sum((x - mu_v) ** 2 for x in log_vols) / len(log_vols)
|
||
sigma_v = math.sqrt(var_v)
|
||
if sigma_v > 0.0:
|
||
z_v = (log_v_t - mu_v) / sigma_v
|
||
|
||
m_regime = 1.0 + config.regime_return_weight * abs(z_r) + config.regime_volume_weight * abs(z_v)
|
||
# Guard against NaN propagation from upstream data
|
||
if math.isnan(m_regime) or math.isinf(m_regime):
|
||
return 1.0
|
||
return max(1.0, min(m_regime, config.regime_multiplier_max))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Combined document signal weight
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass
|
||
class SignalWeight:
|
||
"""Breakdown of a document's aggregation weight."""
|
||
|
||
recency: float
|
||
credibility: float
|
||
novelty_bonus: float
|
||
confidence_gate: float # 0.0 or 1.0
|
||
market_ctx_multiplier: float # >= 1.0
|
||
combined: float
|
||
|
||
# New optional fields for probabilistic mode
|
||
sigmoid_gate: float | None = None # Smooth gate value [0, 1]
|
||
info_gain_factor: float = 1.0 # Surprise multiplier
|
||
source_accuracy_factor: float = 1.0 # Historical accuracy multiplier
|
||
regime_multiplier: float | None = None # M_regime replacing M_context
|
||
|
||
|
||
def compute_signal_weight(
|
||
published_at: datetime,
|
||
reference_time: datetime,
|
||
window: str,
|
||
source_credibility: float,
|
||
novelty_score: float = 0.5,
|
||
extraction_confidence: float = 0.5,
|
||
market_ctx: MarketContext | None = None,
|
||
config: ScoringConfig = DEFAULT_CONFIG,
|
||
*,
|
||
event_type: str | None = None,
|
||
impact_score: float = 0.5,
|
||
source_accuracy_factor: float = 1.0,
|
||
returns: list[float] | None = None,
|
||
volumes: list[float] | None = None,
|
||
) -> SignalWeight:
|
||
"""Compute the combined aggregation weight for a single document signal.
|
||
|
||
When ``config.probabilistic`` is False (default), the formula is:
|
||
combined = confidence_gate * recency * credibility
|
||
* (1 + novelty_bonus) * market_ctx_multiplier
|
||
|
||
When ``config.probabilistic`` is True, the formula is:
|
||
combined = sigmoid_gate * recency(adaptive) * credibility
|
||
* (1 + novelty_bonus) * info_gain * source_accuracy
|
||
* regime_multiplier
|
||
|
||
Args:
|
||
published_at: Document publication time.
|
||
reference_time: Aggregation anchor time.
|
||
window: Trend window identifier.
|
||
source_credibility: Source credibility score (0-1).
|
||
novelty_score: Document novelty score (0-1).
|
||
extraction_confidence: Extraction confidence from the model (0-1).
|
||
market_ctx: Optional market context features for the symbol.
|
||
config: Scoring parameters.
|
||
event_type: Optional event type for information gain computation.
|
||
impact_score: Signal impact score in [0, 1] (default 0.5).
|
||
source_accuracy_factor: Historical source accuracy factor (default 1.0).
|
||
returns: Optional list of recent daily returns for regime multiplier.
|
||
volumes: Optional list of recent daily volumes for regime multiplier.
|
||
|
||
Returns:
|
||
A ``SignalWeight`` with the component breakdown and combined score.
|
||
"""
|
||
cred = credibility_weight(source_credibility, config)
|
||
bonus = novelty_score * config.novelty_bonus_max
|
||
|
||
if not config.probabilistic:
|
||
# --- Heuristic mode: preserve exact current formula ---
|
||
gate = 1.0 if extraction_confidence >= config.confidence_floor else 0.0
|
||
rec = recency_weight(published_at, reference_time, window, config)
|
||
mkt_mult = market_context_multiplier(market_ctx, config)
|
||
|
||
combined = gate * rec * cred * (1.0 + bonus) * mkt_mult
|
||
|
||
return SignalWeight(
|
||
recency=rec,
|
||
credibility=cred,
|
||
novelty_bonus=bonus,
|
||
confidence_gate=gate,
|
||
market_ctx_multiplier=mkt_mult,
|
||
combined=combined,
|
||
)
|
||
|
||
# --- Probabilistic mode ---
|
||
|
||
# 1. Sigmoid confidence gate (Req 2.1–2.5)
|
||
sg = sigmoid_gate(extraction_confidence, config.sigmoid_steepness, config.sigmoid_midpoint)
|
||
|
||
# 2. Information gain factor (Req 3.1–3.5)
|
||
ig = compute_info_gain(
|
||
event_type,
|
||
lambda_param=config.info_gain_lambda,
|
||
max_gain=config.info_gain_max,
|
||
default_base_rate=config.default_base_rate,
|
||
)
|
||
|
||
# 3. Regime multiplier (Req 6.1–6.5) — replaces market_context_multiplier
|
||
rm = compute_regime_multiplier(returns, volumes, config)
|
||
|
||
# 4. Adaptive recency decay (Req 5.1–5.7)
|
||
base_half_life = config.half_life_hours.get(window, 72.0)
|
||
adaptive_hl = compute_adaptive_half_life(
|
||
base_half_life=base_half_life,
|
||
impact_score=impact_score,
|
||
info_gain_factor=ig,
|
||
market_multiplier=rm,
|
||
config=config,
|
||
)
|
||
rec = recency_weight(
|
||
published_at, reference_time, window, config,
|
||
half_life_override=adaptive_hl,
|
||
)
|
||
|
||
# 5. Source accuracy factor (Req 4.2–4.3)
|
||
saf = source_accuracy_factor
|
||
|
||
# 6. Combined weight
|
||
combined = sg * rec * cred * (1.0 + bonus) * ig * saf * rm
|
||
|
||
return SignalWeight(
|
||
recency=rec,
|
||
credibility=cred,
|
||
novelty_bonus=bonus,
|
||
confidence_gate=sg, # sigmoid gate value in probabilistic mode
|
||
market_ctx_multiplier=rm, # regime multiplier stored here for compat
|
||
combined=combined,
|
||
sigmoid_gate=sg,
|
||
info_gain_factor=ig,
|
||
source_accuracy_factor=saf,
|
||
regime_multiplier=rm,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Batch helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass
|
||
class WeightedSignal:
|
||
"""A document intelligence reference paired with its computed weight."""
|
||
|
||
document_id: str
|
||
weight: SignalWeight
|
||
sentiment_value: float # numeric sentiment: +1 positive, -1 negative, 0 neutral/mixed
|
||
impact_score: float
|
||
|
||
# New optional fields for probabilistic mode
|
||
info_gain_factor: float = 1.0 # r = 1 + λ·(-log₂ P(event_type))
|
||
source_accuracy_factor: float = 1.0 # [0.5, 1.5] from historical accuracy
|
||
adaptive_half_life: float | None = None # τ_i when adaptive decay is active
|
||
|
||
|
||
def sentiment_to_numeric(sentiment: str) -> float:
|
||
"""Map a sentiment label to a signed numeric value."""
|
||
mapping = {
|
||
"positive": 1.0,
|
||
"negative": -1.0,
|
||
"neutral": 0.0,
|
||
"mixed": 0.0,
|
||
}
|
||
return mapping.get(sentiment.lower(), 0.0)
|
||
|
||
|
||
async def normalize_impact_scores(
|
||
pool: asyncpg.Pool,
|
||
ticker: str,
|
||
raw_scores: list[float],
|
||
) -> list[float]:
|
||
"""Normalize impact scores using 7-day rolling z-score per ticker.
|
||
|
||
Computes the 7-day mean and stddev of impact_score values from
|
||
document_impact_records for the given ticker, then normalizes
|
||
each raw score: normalized = (raw - mean_7d) / max(stddev_7d, 0.1)
|
||
|
||
The 0.1 floor prevents division by near-zero stddev for low-activity tickers.
|
||
|
||
Fallback: if fewer than 5 records in the 7-day window, returns raw scores
|
||
unchanged (insufficient data for meaningful normalization).
|
||
|
||
Args:
|
||
pool: asyncpg connection pool.
|
||
ticker: The ticker symbol to query distribution for.
|
||
raw_scores: List of raw impact_score values to normalize.
|
||
|
||
Returns:
|
||
List of normalized impact scores (same length as input).
|
||
"""
|
||
if not raw_scores:
|
||
return []
|
||
|
||
row = await pool.fetchrow(
|
||
"""
|
||
SELECT AVG(impact_score) as mean,
|
||
STDDEV(impact_score) as stddev,
|
||
COUNT(*) as cnt
|
||
FROM document_impact_records
|
||
WHERE ticker = $1 AND created_at >= NOW() - INTERVAL '7 days'
|
||
""",
|
||
ticker,
|
||
)
|
||
|
||
# Fallback: insufficient data for meaningful normalization
|
||
if row is None or row["cnt"] < 5:
|
||
return list(raw_scores)
|
||
|
||
mean_7d: float = float(row["mean"])
|
||
stddev_7d: float = float(row["stddev"]) if row["stddev"] is not None and row["stddev"] > 0 else 0.0
|
||
|
||
# Apply 0.1 floor to prevent division by near-zero stddev
|
||
effective_stddev = max(stddev_7d, 0.1)
|
||
|
||
return [(raw - mean_7d) / effective_stddev for raw in raw_scores]
|
||
|
||
|
||
def weighted_sentiment_average(signals: list[WeightedSignal]) -> float:
|
||
"""Compute a weight-adjusted average sentiment across signals.
|
||
|
||
Returns a value in [-1, 1]. Returns 0.0 when total weight is zero.
|
||
"""
|
||
total_weight = 0.0
|
||
weighted_sum = 0.0
|
||
for sig in signals:
|
||
w = sig.weight.combined * sig.impact_score
|
||
weighted_sum += w * sig.sentiment_value
|
||
total_weight += w
|
||
if total_weight == 0.0:
|
||
return 0.0
|
||
return weighted_sum / total_weight
|
||
|
||
|
||
# ===========================================================================
|
||
# V3 Calibrated Evidence Engine — EvidenceUnit and Normalization
|
||
# ===========================================================================
|
||
# All code below this line implements the v3 pipeline. It is gated behind
|
||
# the `v3_engine_enabled` feature flag at the worker/orchestration layer.
|
||
# ===========================================================================
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# V3 Event type base rates (expanded for v3 pipeline)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
V3_EVENT_TYPE_BASE_RATES: dict[str, float] = {
|
||
"earnings": 0.25,
|
||
"guidance": 0.20,
|
||
"merger_acquisition": 0.05,
|
||
"product_launch": 0.15,
|
||
"regulatory": 0.10,
|
||
"management_change": 0.08,
|
||
"partnership": 0.12,
|
||
"legal": 0.07,
|
||
"analyst_rating": 0.30,
|
||
"market_data": 0.40,
|
||
}
|
||
V3_DEFAULT_BASE_RATE: float = 0.10
|
||
|
||
# Direction mapping constants
|
||
_POSITIVE_DIRECTIONS: frozenset[str] = frozenset({"positive", "bullish"})
|
||
_NEGATIVE_DIRECTIONS: frozenset[str] = frozenset({"negative", "bearish"})
|
||
_NEUTRAL_DIRECTIONS: frozenset[str] = frozenset({"neutral", "mixed"})
|
||
|
||
# Macro horizon mapping
|
||
_MACRO_HORIZON_MAP: dict[str, str] = {
|
||
"short_term": "7d",
|
||
"medium_term": "30d",
|
||
"long_term": "90d",
|
||
}
|
||
|
||
# Valid horizons
|
||
_VALID_HORIZONS: frozenset[str] = frozenset({"intraday", "1d", "7d", "30d", "90d"})
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# EvidenceUnit dataclass
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class EvidenceUnit:
|
||
"""Canonical normalized signal representation for the v3 pipeline.
|
||
|
||
Every signal — company, macro, or competitive — is normalized into this
|
||
shape before entering the calibrated reliability / LLR pipeline.
|
||
"""
|
||
|
||
symbol: str
|
||
layer: str # "company" | "macro" | "competitive"
|
||
event_type: str
|
||
source_id: str
|
||
source_group: str
|
||
timestamp: datetime
|
||
horizon: str # "intraday" | "1d" | "7d" | "30d" | "90d"
|
||
direction: int # -1, 0, +1
|
||
sentiment_strength: float # [0, 1]
|
||
impact: float # [0, 1]
|
||
extraction_conf: float # [0, 1]
|
||
source_cred: float # [0, 1]
|
||
novelty: float # [0, 1]
|
||
event_base_rate: float # (0, 1]
|
||
cluster_id: str
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helper functions
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _map_direction(direction_str: str | None) -> int:
|
||
"""Map a sentiment/impact_direction string to a numeric direction.
|
||
|
||
Returns:
|
||
+1 for positive/bullish, -1 for negative/bearish, 0 for neutral/mixed/unknown.
|
||
"""
|
||
if direction_str is None:
|
||
return 0
|
||
lowered = direction_str.lower().strip()
|
||
if lowered in _POSITIVE_DIRECTIONS:
|
||
return 1
|
||
if lowered in _NEGATIVE_DIRECTIONS:
|
||
return -1
|
||
return 0
|
||
|
||
|
||
def _get_event_base_rate(event_type: str | None) -> float:
|
||
"""Look up base rate for an event type, defaulting to 0.10."""
|
||
if event_type is None:
|
||
return V3_DEFAULT_BASE_RATE
|
||
return V3_EVENT_TYPE_BASE_RATES.get(event_type, V3_DEFAULT_BASE_RATE)
|
||
|
||
|
||
def _compute_cluster_id(
|
||
symbol: str,
|
||
horizon: str,
|
||
event_type: str,
|
||
source_group: str,
|
||
time_bucket: str,
|
||
) -> str:
|
||
"""Compute a deterministic cluster_id from the grouping key."""
|
||
key = f"{symbol}|{horizon}|{event_type}|{source_group}|{time_bucket}"
|
||
return hashlib.sha256(key.encode()).hexdigest()[:16]
|
||
|
||
|
||
def _default_time_bucket(ts: datetime, horizon: str) -> str:
|
||
"""Compute a time bucket string for clustering based on horizon.
|
||
|
||
Bucket resolution per horizon:
|
||
intraday → 1h, 1d → 4h, 7d → 24h, 30d → 72h, 90d → 168h
|
||
"""
|
||
bucket_hours: dict[str, int] = {
|
||
"intraday": 1,
|
||
"1d": 4,
|
||
"7d": 24,
|
||
"30d": 72,
|
||
"90d": 168,
|
||
}
|
||
hours = bucket_hours.get(horizon, 24)
|
||
# Truncate timestamp to bucket boundary
|
||
epoch_hours = int(ts.timestamp() / 3600)
|
||
bucket_start = (epoch_hours // hours) * hours
|
||
return str(bucket_start)
|
||
|
||
|
||
def _safe_float(value: Any, default: float = 0.5) -> float:
|
||
"""Extract a float value, substituting default for missing/None."""
|
||
if value is None:
|
||
return default
|
||
try:
|
||
return float(value)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def _clamp(value: float, lo: float, hi: float) -> float:
|
||
"""Clamp a value to [lo, hi]."""
|
||
return max(lo, min(value, hi))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Normalization functions
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def normalize_company_signal(
|
||
signal: dict[str, Any],
|
||
*,
|
||
cluster_id: str | None = None,
|
||
) -> EvidenceUnit | None:
|
||
"""Normalize a company signal into an EvidenceUnit.
|
||
|
||
Args:
|
||
signal: Dict with keys from document_impact_records or similar.
|
||
Required: symbol, timestamp, source_id
|
||
Optional: event_type, source_group, horizon, sentiment,
|
||
sentiment_strength, impact, extraction_conf,
|
||
source_cred, novelty
|
||
cluster_id: If provided, use this cluster_id. Otherwise compute
|
||
from the signal's grouping key.
|
||
|
||
Returns:
|
||
EvidenceUnit or None if required fields are missing.
|
||
"""
|
||
# Validate required fields
|
||
symbol = signal.get("symbol")
|
||
timestamp = signal.get("timestamp")
|
||
source_id = signal.get("source_id")
|
||
|
||
if not symbol:
|
||
logger.warning("v3: Rejecting company signal — missing 'symbol'. source: %s", signal.get("source_id", "unknown"))
|
||
return None
|
||
if timestamp is None:
|
||
logger.warning("v3: Rejecting company signal — missing 'timestamp'. symbol=%s, source_id=%s", symbol, source_id)
|
||
return None
|
||
if not source_id:
|
||
logger.warning("v3: Rejecting company signal — missing 'source_id'. symbol=%s", symbol)
|
||
return None
|
||
|
||
# Ensure timestamp is datetime
|
||
if isinstance(timestamp, str):
|
||
timestamp = datetime.fromisoformat(timestamp)
|
||
if timestamp.tzinfo is None:
|
||
timestamp = timestamp.replace(tzinfo=timezone.utc)
|
||
|
||
# Extract and default fields
|
||
event_type = signal.get("event_type") or "unknown"
|
||
source_group = signal.get("source_group") or "company"
|
||
horizon = signal.get("horizon") or "7d"
|
||
if horizon not in _VALID_HORIZONS:
|
||
horizon = "7d"
|
||
|
||
# Direction mapping
|
||
direction = _map_direction(signal.get("sentiment") or signal.get("direction"))
|
||
|
||
# Optional numeric fields — default to 0.5 if missing
|
||
sentiment_strength = _clamp(_safe_float(signal.get("sentiment_strength")), 0.0, 1.0)
|
||
impact = _clamp(_safe_float(signal.get("impact")), 0.0, 1.0)
|
||
extraction_conf = _clamp(_safe_float(signal.get("extraction_conf") or signal.get("extraction_confidence")), 0.0, 1.0)
|
||
source_cred = _clamp(_safe_float(signal.get("source_cred") or signal.get("source_credibility")), 0.0, 1.0)
|
||
novelty = _clamp(_safe_float(signal.get("novelty") or signal.get("novelty_score")), 0.0, 1.0)
|
||
|
||
# Event base rate
|
||
event_base_rate = _get_event_base_rate(event_type)
|
||
|
||
# Cluster ID
|
||
if cluster_id is None:
|
||
time_bucket = _default_time_bucket(timestamp, horizon)
|
||
cluster_id = _compute_cluster_id(symbol, horizon, event_type, source_group, time_bucket)
|
||
|
||
return EvidenceUnit(
|
||
symbol=str(symbol),
|
||
layer="company",
|
||
event_type=event_type,
|
||
source_id=str(source_id),
|
||
source_group=source_group,
|
||
timestamp=timestamp,
|
||
horizon=horizon,
|
||
direction=direction,
|
||
sentiment_strength=sentiment_strength,
|
||
impact=impact,
|
||
extraction_conf=extraction_conf,
|
||
source_cred=source_cred,
|
||
novelty=novelty,
|
||
event_base_rate=event_base_rate,
|
||
cluster_id=cluster_id,
|
||
)
|
||
|
||
|
||
def normalize_macro_signal(
|
||
signal: dict[str, Any],
|
||
*,
|
||
cluster_id: str | None = None,
|
||
) -> EvidenceUnit | None:
|
||
"""Normalize a macro signal into an EvidenceUnit.
|
||
|
||
Macro signals come from macro_impact_records joined with global_events.
|
||
|
||
Args:
|
||
signal: Dict with keys from macro impact/global event records.
|
||
Required: symbol (or ticker), timestamp, source_id (or event_id)
|
||
Optional: event_type, impact_direction, macro_impact_score,
|
||
event_confidence, estimated_duration, novelty
|
||
cluster_id: If provided, use this cluster_id.
|
||
|
||
Returns:
|
||
EvidenceUnit or None if required fields are missing.
|
||
"""
|
||
# Validate required fields
|
||
symbol = signal.get("symbol") or signal.get("ticker")
|
||
timestamp = signal.get("timestamp")
|
||
source_id = signal.get("source_id") or signal.get("event_id")
|
||
|
||
if not symbol:
|
||
logger.warning("v3: Rejecting macro signal — missing 'symbol'/'ticker'. source: %s", signal.get("source_id", "unknown"))
|
||
return None
|
||
if timestamp is None:
|
||
logger.warning("v3: Rejecting macro signal — missing 'timestamp'. symbol=%s, source_id=%s", symbol, source_id)
|
||
return None
|
||
if not source_id:
|
||
logger.warning("v3: Rejecting macro signal — missing 'source_id'/'event_id'. symbol=%s", symbol)
|
||
return None
|
||
|
||
# Ensure timestamp is datetime
|
||
if isinstance(timestamp, str):
|
||
timestamp = datetime.fromisoformat(timestamp)
|
||
if timestamp.tzinfo is None:
|
||
timestamp = timestamp.replace(tzinfo=timezone.utc)
|
||
|
||
# Extract fields
|
||
event_type = signal.get("event_type") or "unknown"
|
||
source_group = "macro"
|
||
|
||
# Horizon from estimated_duration
|
||
estimated_duration = signal.get("estimated_duration") or "medium_term"
|
||
horizon = _MACRO_HORIZON_MAP.get(estimated_duration, "30d")
|
||
|
||
# Direction from impact_direction
|
||
direction = _map_direction(signal.get("impact_direction") or signal.get("direction"))
|
||
|
||
# Impact from macro_impact_score
|
||
impact = _clamp(_safe_float(signal.get("macro_impact_score") or signal.get("impact")), 0.0, 1.0)
|
||
|
||
# Source cred and extraction conf from event_confidence
|
||
event_confidence = _safe_float(signal.get("event_confidence") or signal.get("confidence"))
|
||
source_cred = _clamp(event_confidence, 0.0, 1.0)
|
||
extraction_conf = _clamp(event_confidence, 0.0, 1.0)
|
||
|
||
# Novelty: 1.0 for new events (as per requirement 1.2)
|
||
novelty = _clamp(_safe_float(signal.get("novelty"), default=1.0), 0.0, 1.0)
|
||
|
||
# Sentiment strength — default 0.5 for macro
|
||
sentiment_strength = _clamp(_safe_float(signal.get("sentiment_strength")), 0.0, 1.0)
|
||
|
||
# Event base rate
|
||
event_base_rate = _get_event_base_rate(event_type)
|
||
|
||
# Cluster ID
|
||
if cluster_id is None:
|
||
time_bucket = _default_time_bucket(timestamp, horizon)
|
||
cluster_id = _compute_cluster_id(str(symbol), horizon, event_type, source_group, time_bucket)
|
||
|
||
return EvidenceUnit(
|
||
symbol=str(symbol),
|
||
layer="macro",
|
||
event_type=event_type,
|
||
source_id=str(source_id),
|
||
source_group=source_group,
|
||
timestamp=timestamp,
|
||
horizon=horizon,
|
||
direction=direction,
|
||
sentiment_strength=sentiment_strength,
|
||
impact=impact,
|
||
extraction_conf=extraction_conf,
|
||
source_cred=source_cred,
|
||
novelty=novelty,
|
||
event_base_rate=event_base_rate,
|
||
cluster_id=cluster_id,
|
||
)
|
||
|
||
|
||
def normalize_competitive_signal(
|
||
signal: dict[str, Any],
|
||
*,
|
||
cluster_id: str | None = None,
|
||
) -> EvidenceUnit | None:
|
||
"""Normalize a competitive signal into an EvidenceUnit.
|
||
|
||
Competitive signals come from pattern mining and cross-company propagation.
|
||
|
||
Args:
|
||
signal: Dict with keys from competitive_signal_records.
|
||
Required: symbol (or target_ticker), timestamp, source_id (or source_document_id)
|
||
Optional: event_type, signal_direction, signal_strength,
|
||
relationship_strength, pattern_confidence, time_horizon
|
||
cluster_id: If provided, use this cluster_id.
|
||
|
||
Returns:
|
||
EvidenceUnit or None if required fields are missing.
|
||
"""
|
||
# Validate required fields
|
||
symbol = signal.get("symbol") or signal.get("target_ticker")
|
||
timestamp = signal.get("timestamp")
|
||
source_id = signal.get("source_id") or signal.get("source_document_id")
|
||
|
||
if not symbol:
|
||
logger.warning("v3: Rejecting competitive signal — missing 'symbol'/'target_ticker'. source: %s", signal.get("source_id", "unknown"))
|
||
return None
|
||
if timestamp is None:
|
||
logger.warning("v3: Rejecting competitive signal — missing 'timestamp'. symbol=%s, source_id=%s", symbol, source_id)
|
||
return None
|
||
if not source_id:
|
||
logger.warning("v3: Rejecting competitive signal — missing 'source_id'/'source_document_id'. symbol=%s", symbol)
|
||
return None
|
||
|
||
# Ensure timestamp is datetime
|
||
if isinstance(timestamp, str):
|
||
timestamp = datetime.fromisoformat(timestamp)
|
||
if timestamp.tzinfo is None:
|
||
timestamp = timestamp.replace(tzinfo=timezone.utc)
|
||
|
||
# Extract fields
|
||
event_type = signal.get("event_type") or "unknown"
|
||
source_group = "competitive"
|
||
|
||
# Horizon from time_horizon field
|
||
time_horizon = signal.get("time_horizon") or signal.get("horizon") or "7d"
|
||
if time_horizon in _MACRO_HORIZON_MAP:
|
||
horizon = _MACRO_HORIZON_MAP[time_horizon]
|
||
elif time_horizon in _VALID_HORIZONS:
|
||
horizon = time_horizon
|
||
else:
|
||
horizon = "7d"
|
||
|
||
# Direction from signal_direction (bullish/bearish/neutral)
|
||
direction = _map_direction(signal.get("signal_direction") or signal.get("direction"))
|
||
|
||
# Impact = signal_strength × relationship_strength (Req 1.3)
|
||
signal_strength = _safe_float(signal.get("signal_strength"))
|
||
relationship_strength = _safe_float(signal.get("relationship_strength"))
|
||
impact = _clamp(signal_strength * relationship_strength, 0.0, 1.0)
|
||
|
||
# Source cred from pattern_confidence (Req 1.3)
|
||
pattern_confidence = _safe_float(signal.get("pattern_confidence"))
|
||
source_cred = _clamp(pattern_confidence, 0.0, 1.0)
|
||
|
||
# Extraction conf = pattern_confidence (Req 1.3)
|
||
extraction_conf = _clamp(pattern_confidence, 0.0, 1.0)
|
||
|
||
# Novelty: 1.0 for competitive signals (Req 1.3)
|
||
novelty = _clamp(_safe_float(signal.get("novelty"), default=1.0), 0.0, 1.0)
|
||
|
||
# Sentiment strength — default 0.5 for competitive
|
||
sentiment_strength = _clamp(_safe_float(signal.get("sentiment_strength")), 0.0, 1.0)
|
||
|
||
# Event base rate
|
||
event_base_rate = _get_event_base_rate(event_type)
|
||
|
||
# Cluster ID
|
||
if cluster_id is None:
|
||
time_bucket = _default_time_bucket(timestamp, horizon)
|
||
cluster_id = _compute_cluster_id(str(symbol), horizon, event_type, source_group, time_bucket)
|
||
|
||
return EvidenceUnit(
|
||
symbol=str(symbol),
|
||
layer="competitive",
|
||
event_type=event_type,
|
||
source_id=str(source_id),
|
||
source_group=source_group,
|
||
timestamp=timestamp,
|
||
horizon=horizon,
|
||
direction=direction,
|
||
sentiment_strength=sentiment_strength,
|
||
impact=impact,
|
||
extraction_conf=extraction_conf,
|
||
source_cred=source_cred,
|
||
novelty=novelty,
|
||
event_base_rate=event_base_rate,
|
||
cluster_id=cluster_id,
|
||
)
|
||
|
||
|
||
# ===========================================================================
|
||
# V3 Calibrated Reliability Pipeline
|
||
# ===========================================================================
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SourceStats:
|
||
"""Historical accuracy stats for a signal source (Bayesian prior).
|
||
|
||
Used to compute q_source via Beta-Binomial shrinkage.
|
||
"""
|
||
|
||
source_id: str
|
||
hits: int = 0 # correct directional predictions
|
||
misses: int = 0 # incorrect directional predictions
|
||
alpha_0: float = 3.0 # Beta prior alpha (pseudo-successes)
|
||
beta_0: float = 3.0 # Beta prior beta (pseudo-failures)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ReliabilityComponents:
|
||
"""Breakdown of calibrated reliability for a single signal.
|
||
|
||
Each q_* factor is in [0, 1] and represents one quality dimension.
|
||
q_i is the final combined reliability used downstream in LLR conversion.
|
||
"""
|
||
|
||
q_ext: float # extraction confidence reliability
|
||
q_source: float # source accuracy reliability (Bayesian shrinkage)
|
||
q_recency: float # temporal freshness reliability
|
||
q_uniqueness: float # novelty / de-duplication reliability
|
||
q_i: float # final combined: clamp(q_ext × q_source × source_cred × q_recency × q_uniqueness, 0, 1)
|
||
|
||
|
||
# Horizon-specific base half-lives for recency decay (hours)
|
||
_V3_TAU_BASE: dict[str, float] = {
|
||
"intraday": 2.0,
|
||
"1d": 12.0,
|
||
"7d": 72.0,
|
||
"30d": 240.0,
|
||
"90d": 720.0,
|
||
}
|
||
|
||
|
||
def _sigmoid(x: float) -> float:
|
||
"""Compute sigmoid(x) = 1 / (1 + exp(-x)) with overflow guard."""
|
||
if x < -500.0:
|
||
return 0.0
|
||
if x > 500.0:
|
||
return 1.0
|
||
return 1.0 / (1.0 + math.exp(-x))
|
||
|
||
|
||
def compute_v3_reliability(
|
||
unit: EvidenceUnit,
|
||
source_stats: SourceStats,
|
||
cluster_position: int, # duplicate_count_before
|
||
reference_time: datetime,
|
||
) -> ReliabilityComponents:
|
||
"""Compute calibrated reliability components for an EvidenceUnit.
|
||
|
||
Implements Requirements 2.1–2.9: extraction confidence gate, Bayesian
|
||
source accuracy, adaptive recency decay, and novelty/uniqueness penalty.
|
||
|
||
Args:
|
||
unit: The normalized evidence unit to score.
|
||
source_stats: Historical accuracy record for the signal's source.
|
||
cluster_position: Number of signals in the same cluster ingested
|
||
before this one (duplicate_count_before). 0 for first-in-cluster.
|
||
reference_time: The "now" anchor for computing age_hours.
|
||
|
||
Returns:
|
||
ReliabilityComponents with individual factors and combined q_i.
|
||
"""
|
||
# --- q_ext: extraction confidence reliability (Req 2.1) ---
|
||
# sigmoid(8.0 × (extraction_conf - 0.55))
|
||
q_ext = _sigmoid(8.0 * (unit.extraction_conf - 0.55))
|
||
|
||
# --- q_source: Bayesian shrinkage source reliability (Req 2.2, 2.3) ---
|
||
alpha = source_stats.alpha_0 + source_stats.hits
|
||
beta = source_stats.beta_0 + source_stats.misses
|
||
e_theta = alpha / (alpha + beta)
|
||
# clamp((E[theta] - 0.50) / 0.35, 0, 1)
|
||
q_source = _clamp((e_theta - 0.50) / 0.35, 0.0, 1.0)
|
||
|
||
# --- q_recency: adaptive exponential decay (Req 2.4, 2.5, 2.6) ---
|
||
# Ensure tz-aware timestamps
|
||
ts = unit.timestamp
|
||
if ts.tzinfo is None:
|
||
ts = ts.replace(tzinfo=timezone.utc)
|
||
ref = reference_time
|
||
if ref.tzinfo is None:
|
||
ref = ref.replace(tzinfo=timezone.utc)
|
||
|
||
age_hours = max((ref - ts).total_seconds() / 3600.0, 0.0)
|
||
|
||
# Adaptive half-life: tau_adaptive = tau_base × (1 + 0.75 × impact + 0.50 × surprise)
|
||
# surprise = clamp(-log2(event_base_rate) / 5, 0, 1)
|
||
event_base_rate = unit.event_base_rate
|
||
if event_base_rate <= 0.0:
|
||
event_base_rate = 0.10 # Req 2.5: default to 0.10 to prevent log(0)
|
||
|
||
surprise = _clamp(-math.log2(event_base_rate) / 5.0, 0.0, 1.0)
|
||
|
||
tau_base = _V3_TAU_BASE.get(unit.horizon, 72.0)
|
||
tau_adaptive = tau_base * (1.0 + 0.75 * unit.impact + 0.50 * surprise)
|
||
|
||
# q_recency = 2^(-age_hours / tau_adaptive)
|
||
# Guard against extreme exponents
|
||
if tau_adaptive <= 0.0:
|
||
tau_adaptive = tau_base # fallback
|
||
exponent = -age_hours / tau_adaptive
|
||
# For very large negative exponents, result is effectively 0
|
||
if exponent < -1000.0:
|
||
q_recency = 0.0
|
||
else:
|
||
q_recency = math.pow(2.0, exponent)
|
||
|
||
# --- q_uniqueness: novelty + de-duplication (Req 2.7) ---
|
||
# clamp(0.5 + 0.5 × novelty, 0.5, 1.0) × (1 / sqrt(1 + dup_count))
|
||
novelty_factor = _clamp(0.5 + 0.5 * unit.novelty, 0.5, 1.0)
|
||
dedup_factor = 1.0 / math.sqrt(1.0 + cluster_position)
|
||
q_uniqueness = novelty_factor * dedup_factor
|
||
|
||
# --- q_i: combined reliability (Req 2.8) ---
|
||
q_i = _clamp(
|
||
q_ext * q_source * unit.source_cred * q_recency * q_uniqueness,
|
||
0.0,
|
||
1.0,
|
||
)
|
||
|
||
# --- Explainability floor on q_recency (Req 2.9) ---
|
||
# Apply floor of 0.01 only for the display value; q_i uses raw q_recency
|
||
q_recency_display = max(q_recency, 0.01)
|
||
|
||
return ReliabilityComponents(
|
||
q_ext=q_ext,
|
||
q_source=q_source,
|
||
q_recency=q_recency_display,
|
||
q_uniqueness=q_uniqueness,
|
||
q_i=q_i,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# V3 LLR Conversion (Requirements 3.1–3.6)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def compute_llr(unit: EvidenceUnit, q_i: float) -> float:
|
||
"""Convert calibrated reliability to log-likelihood ratio.
|
||
|
||
Requirements: 3.1–3.6
|
||
|
||
Formula:
|
||
p_correct = clamp(0.50 + 0.35 × q_i × impact × sentiment_strength, 0.501, 0.85)
|
||
LLR_i = direction × ln(p_correct / (1 - p_correct))
|
||
|
||
For neutral signals (direction == 0), returns 0.0 immediately.
|
||
For directional signals, the LLR sign always matches direction.
|
||
|
||
Bounds:
|
||
- Minimum |LLR| ≈ ln(0.501/0.499) ≈ 0.004 for directional signals
|
||
- Maximum |LLR| ≈ ln(0.85/0.15) ≈ 1.735
|
||
|
||
Args:
|
||
unit: The normalized evidence unit containing direction, impact,
|
||
and sentiment_strength.
|
||
q_i: The combined calibrated reliability from compute_v3_reliability.
|
||
|
||
Returns:
|
||
Log-likelihood ratio. Positive for bullish, negative for bearish,
|
||
zero for neutral.
|
||
"""
|
||
# Req 3.5: Neutral signals produce zero LLR
|
||
if unit.direction == 0:
|
||
return 0.0
|
||
|
||
# Req 3.1–3.2: Compute p_correct with calibrated reliability
|
||
p_correct = _clamp(
|
||
0.50 + 0.35 * q_i * unit.impact * unit.sentiment_strength,
|
||
0.501,
|
||
0.85,
|
||
)
|
||
|
||
# Req 3.3–3.4: LLR_i = direction × ln(p_correct / (1 - p_correct))
|
||
llr = unit.direction * math.log(p_correct / (1.0 - p_correct))
|
||
|
||
return llr
|