60 lines
1.8 KiB
Python
60 lines
1.8 KiB
Python
"""News-driven strategy — trade on aggregated news sentiment."""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from shared.schemas.trading import MarketSnapshot, SentimentContext, SignalDirection, TradeSignal
|
|
from shared.strategies.base import BaseStrategy
|
|
|
|
|
|
class NewsDrivenStrategy(BaseStrategy):
|
|
"""Generate signals from aggregated news sentiment for a ticker.
|
|
|
|
**Buy signal** (LONG):
|
|
``avg_score > 0.3`` AND ``avg_confidence > 0.5`` AND
|
|
``article_count >= 2``.
|
|
|
|
**Sell signal** (SHORT):
|
|
``avg_score < -0.3`` AND ``avg_confidence > 0.5`` AND
|
|
``article_count >= 2``.
|
|
|
|
Signal strength = ``abs(avg_score) * avg_confidence``, clamped to
|
|
[0, 1].
|
|
"""
|
|
|
|
name: str = "news_driven"
|
|
|
|
async def evaluate(
|
|
self,
|
|
ticker: str,
|
|
market: MarketSnapshot,
|
|
sentiment: SentimentContext | None = None,
|
|
) -> TradeSignal | None:
|
|
if sentiment is None:
|
|
return None
|
|
|
|
# Require at least 2 articles for statistical confidence.
|
|
if sentiment.article_count < 2:
|
|
return None
|
|
|
|
# Require minimum confidence.
|
|
if sentiment.avg_confidence <= 0.5:
|
|
return None
|
|
|
|
if sentiment.avg_score > 0.3:
|
|
direction = SignalDirection.LONG
|
|
elif sentiment.avg_score < -0.3:
|
|
direction = SignalDirection.SHORT
|
|
else:
|
|
# Sentiment is neutral — no opinion.
|
|
return None
|
|
|
|
raw_strength = abs(sentiment.avg_score) * sentiment.avg_confidence
|
|
strength = max(0.0, min(1.0, raw_strength))
|
|
|
|
return TradeSignal(
|
|
ticker=ticker,
|
|
direction=direction,
|
|
strength=strength,
|
|
strategy_sources=[self.name],
|
|
timestamp=datetime.now(tz=timezone.utc),
|
|
)
|