cryptosignal-sleuth / backend /signal_engine.py
Alexo19's picture
You are an elite full-stack developer and quant engineer.
06d59d5 verified
```python
from typing import Dict, Any
from dataclasses import dataclass
from .models_registry import ModelsRegistry
import random
@dataclass
class SignalComponents:
trend_score: float # -1 to +1
momentum_score: float # -1 to +1
pattern_score: float # -1 to +1
sentiment_score: float # -1 to +1
volatility: float # 0-1
class SignalEngine:
def __init__(self, models: ModelsRegistry):
self.models = models
def generate_signal(self, components: SignalComponents) -> Dict[str, Any]:
"""Combine all analysis components into a final signal"""
# Weighted average of scores
final_score = (
0.4 * components.trend_score +
0.3 * components.momentum_score +
0.2 * components.pattern_score +
0.1 * components.sentiment_score
)
# Determine direction and confidence
if abs(final_score) < 0.2:
direction = "NEUTRAL"
confidence = max(10, int(50 + 50 * abs(final_score) / 0.2))
else:
direction = "LONG" if final_score > 0 else "SHORT"
confidence = int((abs(final_score) - 0.2) / 0.8 * 90) + 10
# Calculate levels based on volatility
base_price = 42000 # In real app, this would come from market data
atr = components.volatility * base_price * 0.05
if direction == "LONG":
entry_min = base_price - atr * 0.5
entry_max = base_price
stop_loss = base_price - atr * 1.5
take_profit = [
base_price + atr * 2,
base_price + atr * 3
]
elif direction == "SHORT":
entry_min = base_price
entry_max = base_price + atr * 0.5
stop_loss = base_price + atr * 1.5
take_profit = [
base_price - atr * 2,
base_price - atr * 3
]
else: # NEUTRAL
entry_min = base_price - atr * 0.2
entry_max = base_price + atr * 0.2
stop_loss = base_price + atr * 2 * (-1 if random.random() > 0.5 else 1)
take_profit = []
# Generate explanation based on components
explanation_parts = []
if components.trend_score > 0.3:
explanation_parts.append("Strong bullish trend detected.")
elif components.trend_score < -0.3:
explanation_parts.append("Strong bearish trend detected.")
if components.momentum_score > 0.4:
explanation_parts.append("Momentum favors upside.")
elif components.momentum_score < -0.4:
explanation_parts.append("Momentum favors downside.")
if components.pattern_score > 0.5:
explanation_parts.append("Bullish chart pattern identified.")
elif components.pattern_score < -0.5:
explanation_parts.append("Bearish chart pattern identified.")
if not explanation_parts:
explanation_parts.append("No strong signals detected - market may be ranging.")
explanation = " ".join(explanation_parts)
return {
"direction": direction,
"entry_zone": [entry_min, entry_max],
"stop_loss": stop_loss,
"take_profit_levels": take_profit,
"confidence": confidence,
"explanation": explanation
}
async def analyze_market(self, pair: str, timeframe: str) -> Dict[str, Any]:
"""Analyze market conditions and generate signal"""
# In a real implementation, we would:
# 1. Fetch market data
# 2. Call time-series models
# 3. Calculate technical indicators
# 4. Get sentiment if available
# Mock analysis for demo purposes
components = SignalComponents(
trend_score=random.uniform(-1, 1),
momentum_score=random.uniform(-0.5, 0.8),
pattern_score=random.uniform(-1, 1),
sentiment_score=random.uniform(-0.3, 0.3),
volatility=random.uniform(0.2, 0.8)
)
return self.generate_signal(components)
async def analyze_screenshot(self, image_data: bytes) -> Dict[str, Any]:
"""Analyze uploaded chart screenshot"""
# In a real implementation, we would:
# 1. Process image with vision model
# 2. Extract text with OCR
# 3. Analyze patterns and generate signal
# Mock analysis for demo purposes
components = SignalComponents(
trend_score=random.uniform(0, 1), # Screenshots more likely to be long ideas
momentum_score=random.uniform(0, 0.9),
pattern_score=random.uniform(0.1, 1),
sentiment_score=random.uniform(0, 0.5),
volatility=random.uniform(0.3, 0.9)
)
return self.generate_signal(components)
```
Let me continue with the frontend setup and other necessary files: