Spaces:
Runtime error
Runtime error
File size: 4,080 Bytes
d5b7ee9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | """Aggregate FinBERT per-headline results into a single symbol-level score.
Supports event-type weighting (earnings/executive/product/macro/generic)
and temporal decay (newer headlines have more impact).
"""
from __future__ import annotations
import time
from datetime import datetime, timezone
from trading_cli.sentiment.news_classifier import EventType, EventClassification, DEFAULT_WEIGHTS
LABEL_DIRECTION = {"positive": 1.0, "negative": -1.0, "neutral": 0.0}
def aggregate_scores(results: list[dict]) -> float:
"""
Weighted average of label directions, weighted by confidence score.
Returns float in [-1.0, +1.0]:
+1.0 = all headlines strongly positive
-1.0 = all headlines strongly negative
0.0 = neutral or empty
"""
if not results:
return 0.0
total_weight = 0.0
weighted_sum = 0.0
for r in results:
label = r.get("label", "neutral")
score = float(r.get("score", 0.5))
direction = LABEL_DIRECTION.get(label, 0.0)
weighted_sum += direction * score
total_weight += score
if total_weight == 0.0:
return 0.0
return max(-1.0, min(1.0, weighted_sum / total_weight))
def aggregate_scores_weighted(
results: list[dict],
classifications: list[EventClassification] | None = None,
timestamps: list[float] | None = None,
event_weights: dict[EventType, float] | None = None,
half_life_hours: float = 24.0,
) -> float:
"""
Weighted sentiment aggregation with event-type and temporal decay.
Args:
results: List of FinBERT results with "label" and "score" keys.
classifications: Optional event classifications for each headline.
timestamps: Optional Unix timestamps for each headline (for temporal decay).
event_weights: Custom event type weight multipliers.
half_life_hours: Hours for temporal half-life decay. Default 24h.
Returns float in [-1.0, +1.0].
"""
if not results:
return 0.0
now = time.time()
total_weight = 0.0
weighted_sum = 0.0
weights = event_weights or DEFAULT_WEIGHTS
for i, r in enumerate(results):
label = r.get("label", "neutral")
score = float(r.get("score", 0.5))
direction = LABEL_DIRECTION.get(label, 0.0)
# Base weight from FinBERT confidence
w = score
# Event type weight multiplier
if classifications and i < len(classifications):
ec = classifications[i]
w *= weights.get(ec.event_type, 1.0)
# Temporal decay: newer headlines weight more
if timestamps and i < len(timestamps):
ts = timestamps[i]
age_hours = (now - ts) / 3600.0
# Exponential decay: weight halves every half_life_hours
decay = 0.5 ** (age_hours / half_life_hours)
w *= decay
weighted_sum += direction * w
total_weight += w
if total_weight == 0.0:
return 0.0
return max(-1.0, min(1.0, weighted_sum / total_weight))
def get_sentiment_summary(results: list[dict]) -> dict:
"""Return counts, dominant label, and aggregate score."""
counts = {"positive": 0, "negative": 0, "neutral": 0}
for r in results:
label = r.get("label", "neutral")
if label in counts:
counts[label] += 1
dominant = max(counts, key=lambda k: counts[k]) if results else "neutral"
return {
"score": aggregate_scores(results),
"positive_count": counts["positive"],
"negative_count": counts["negative"],
"neutral_count": counts["neutral"],
"total": len(results),
"dominant": dominant,
}
def score_to_bar(score: float, width: int = 20) -> str:
"""Render a text gauge like: ──────●────────── for display in terminals."""
clamped = max(-1.0, min(1.0, score))
mid = width // 2
pos = int(mid + clamped * mid)
pos = max(0, min(width - 1, pos))
bar = list("─" * width)
bar[mid] = "┼"
bar[pos] = "●"
return "".join(bar)
|